Merge pull request #37 from sr-lab/terraform_smells

Terraform smells

glitch/__main__.py

@@ -1,6 +1,6 @@
 import click, os, sys
 from glitch.analysis.rules import Error, RuleVisitor
-from glitch.helpers import RulesListOption
+from glitch.helpers import RulesListOption, get_smell_types, get_smells
 from glitch.parsers.docker_parser import DockerParser
 from glitch.stats.print import print_stats
 from glitch.stats.stats import FileStats
@@ -50,12 +50,12 @@ def parse_and_check(type, path, module, parser, analyses, errors, stats):
          "This flag is only relevant if you are using the dataset flag.")
 @click.option('--csv', is_flag=True, default=False,
     help="Use this flag if you want the output to be in CSV format.")
-@click.option('--smells', cls=RulesListOption, multiple=True, 
-    help="The type of smells being analyzed.")
+@click.option('--smell_types', cls=RulesListOption, multiple=True, 
+    help="The type of smell_types being analyzed.")
 @click.argument('path', type=click.Path(exists=True), required=True)
 @click.argument('output', type=click.Path(), required=False)
 def glitch(tech, type, path, config, module, csv, 
-        dataset, includeall, smells, output, tableformat, linter):
+        dataset, includeall, smell_types, output, tableformat, linter):
     if config != "configs/default.ini" and not os.path.exists(config):
         raise click.BadOptionUsage('config', f"Invalid value for 'config': Path '{config}' does not exist.")
     elif os.path.isdir(config):
@@ -74,15 +74,16 @@ def glitch(tech, type, path, config, module, csv,
         parser = DockerParser()
     elif tech == Tech.terraform:
         parser = TerraformParser()
+        config = resource_filename('glitch', "configs/terraform.ini")
     file_stats = FileStats()
 
-    if smells == ():
-        smells = list(map(lambda c: c.get_name(), RuleVisitor.__subclasses__()))
+    if smell_types == ():
+        smell_types = get_smell_types()
 
     analyses = []
     rules = RuleVisitor.__subclasses__()
     for r in rules:
-        if smells == () or r.get_name() in smells:
+        if smell_types == () or r.get_name() in smell_types:
             analysis = r(tech)
             analysis.config(config)
             analyses.append(analysis)
@@ -139,7 +140,7 @@ def glitch(tech, type, path, config, module, csv,
 
     if f != sys.stdout: f.close()
     if not linter:
-        print_stats(errors, smells, file_stats, tableformat)
+        print_stats(errors, get_smells(smell_types, tech), file_stats, tableformat)
 
 def main():
     glitch(prog_name='glitch')

glitch/analysis/design.py

@@ -328,7 +328,7 @@ def check_atomicunit(self, au: AtomicUnit, file: str) -> list[Error]:
     def check_dependency(self, d: Dependency, file: str) -> list[Error]:
         return []
 
-    def check_attribute(self, a: Attribute, file: str) -> list[Error]:
+    def check_attribute(self, a: Attribute, file: str, au: AtomicUnit = None, parent_name: str = "") -> list[Error]:
         return []
 
     def check_variable(self, v: Variable, file: str) -> list[Error]:

glitch/analysis/rules.py

@@ -16,9 +16,32 @@ class Error():
             'sec_invalid_bind': "Invalid IP address binding - Binding to the address 0.0.0.0 allows connections from every possible network which might be a security issues. (CWE-284)",
             'sec_no_int_check': "No integrity check - The content of files downloaded from the internet should be checked. (CWE-353)",
             'sec_no_default_switch': "Missing default case statement - Not handling every possible input combination might allow an attacker to trigger an error for an unhandled value. (CWE-478)",
-            'sec_non_official_image': "Use of non-official Docker image - Use of non-official images should be avoided or taken into careful consideration. (CWE-829)",
             'sec_full_permission_filesystem': "Full permission to the filesystem - Files should not have full permissions to every user. (CWE-732)",
-            'sec_obsolete_command': "Use of obsolete command or function - Avoid using obsolete or deprecated commands and functions. (CWE-477)"
+            'sec_obsolete_command': "Use of obsolete command or function - Avoid using obsolete or deprecated commands and functions. (CWE-477)",
+            Tech.docker: {
+                'sec_non_official_image': "Use of non-official Docker image - Use of non-official images should be avoided or taken into careful consideration. (CWE-829)",
+            },
+            Tech.terraform: {
+                'sec_integrity_policy': "Integrity Policy - Image tag is prone to be mutable or integrity monitoring is disabled. (CWE-471)",
+                'sec_ssl_tls_policy': "SSL/TLS/mTLS Policy - Developers should use SSL/TLS/mTLS protocols and their secure versions. (CWE-326)",
+                'sec_dnssec': "Use of DNS without DNSSEC - Developers should favor the usage of DNSSEC while using DNS. (CWE-350)",
+                'sec_public_ip': "Associated Public IP address - Associating Public IP addresses allows connections from public internet. (CWE-1327)",
+                'sec_access_control': "Insecure Access Control - Developers should be aware of possible unauthorized access. (CWE-284)",
+                'sec_authentication': "Disabled/Weak Authentication - Developers should guarantee that authentication is enabled. (CWE-287 | CWE-306)",
+                'sec_missing_encryption': "Missing Encryption - Developers should ensure encryption of sensitive and critical data. (CWE-311)",
+                'sec_firewall_misconfig': "Firewall Misconfiguration - Developers should favor the usage of a well configured waf. (CWE-693)",
+                'sec_threats_detection_alerts': "Missing Threats Detection/Alerts - Developers should enable threats detection and alerts when it is possible. (CWE-693)",
+                'sec_weak_password_key_policy': "Weak Password/Key Policy - Developers should favor the usage of strong password/key requirements and configurations. (CWE-521).",
+                'sec_sensitive_iam_action': "Sensitive Action by IAM - Developers should use the principle of least privilege when defining IAM policies. (CWE-284)",
+                'sec_key_management': "Key Management - Developers should use well configured Customer Managed Keys (CMK) for encryption. (CWE-1394)",
+                'sec_network_security_rules': "Network Security Rules - Developers should enforce that only secure network rules are being used. (CWE-923)",
+                'sec_permission_iam_policies': "Permission of IAM Policies - Developers should be aware of unwanted permissions of IAM policies. (CWE-732 | CWE-284)",
+                'sec_logging': "Logging - Logs should be enabled and securely configured to help monitoring and preventing security problems. (CWE-223 | CWE-778)",
+                'sec_attached_resource': "Attached Resource - Ensure that Route53 A records point to resources part of your account rather than just random IP addresses. (CWE-200)",
+                'sec_versioning': "Versioning - Ensure that versioning is enabled so that users can retrieve and restore previous versions.",
+                'sec_naming': "Naming - Ensure storage accounts adhere to the naming rules and every security groups and rules have a description. (CWE-1099 | CWE-710)",
+                'sec_replication': "Replication - Ensure that cross-region replication is enabled to allow copying objects across S3 buckets.",
+            }
         },
         'design': {
             'design_imperative_abstraction': "Imperative abstraction - The presence of imperative statements defies the purpose of IaC declarative languages.",
@@ -39,15 +62,20 @@ class Error():
 
     @staticmethod
     def agglomerate_errors():
-        for error_list in Error.ERRORS.values():
-            for k,v in error_list.items():
-                Error.ALL_ERRORS[k] = v
-
-    def __init__(self, code: str, el, path: str, repr: str) -> None:
+        def aux_agglomerate_errors(key, errors):
+            if isinstance(errors, dict):
+                for k, v in errors.items():
+                    aux_agglomerate_errors(k, v)
+            else:
+                Error.ALL_ERRORS[key] = errors
+        aux_agglomerate_errors('', Error.ERRORS)
+
+    def __init__(self, code: str, el, path: str, repr: str, opt_msg: str = None) -> None:
         self.code: str = code
         self.el = el
         self.path = path
         self.repr = repr
+        self.opt_msg = opt_msg
 
         if isinstance(self.el, CodeElement):
             self.line = self.el.line
@@ -56,17 +84,22 @@ def __init__(self, code: str, el, path: str, repr: str) -> None:
 
     def to_csv(self) -> str:
         repr = self.repr.split('\n')[0].strip()
-        return f"{self.path},{self.line},{self.code},{repr}"
+        if self.opt_msg:
+            return f"{self.path},{self.line},{self.code},{repr},{self.opt_msg}"
+        else:
+            return f"{self.path},{self.line},{self.code},{repr},-"
 
     def __repr__(self) -> str:
         with open(self.path) as f:
             line = f.readlines()[self.line - 1].strip() if self.line != -1 else self.repr.split('\n')[0]
+            if self.opt_msg:
+                line += f"\n-> {self.opt_msg}"
             return \
                 f"{self.path}\nIssue on line {self.line}: {Error.ALL_ERRORS[self.code]}\n" + \
                     f"{line}\n"
 
     def __hash__(self):
-        return hash((self.code, self.path, self.line))
+        return hash((self.code, self.path, self.line, self.opt_msg))
 
     def __eq__(self, other):
         if not isinstance(other, type(self)): return NotImplemented
@@ -79,22 +112,24 @@ class RuleVisitor(ABC):
     def __init__(self, tech: Tech) -> None:
         super().__init__()
         self.tech = tech
+        self.code = None
 
     def check(self, code) -> list[Error]:
+        self.code = code
         if isinstance(code, Project):
             return self.check_project(code)
         elif isinstance(code, Module):
             return self.check_module(code)
         elif isinstance(code, UnitBlock):
             return self.check_unitblock(code)
 
-    def check_element(self, c, file: str) -> list[Error]:
+    def check_element(self, c, file: str, au_type = None, parent_name: str = "") -> list[Error]:
         if isinstance(c, AtomicUnit):
             return self.check_atomicunit(c, file)
         elif isinstance(c, Dependency):
             return self.check_dependency(c, file)
         elif isinstance(c, Attribute):
-            return self.check_attribute(c, file)
+            return self.check_attribute(c, file, au_type, parent_name)
         elif isinstance(c, Variable):
             return self.check_variable(c, file)
         elif isinstance(c, ConditionalStatement):
@@ -154,7 +189,7 @@ def check_unitblock(self, u: UnitBlock) -> list[Error]:
     def check_atomicunit(self, au: AtomicUnit, file: str) -> list[Error]:
         errors = []
         for a in au.attributes:
-            errors += self.check_attribute(a, file)
+            errors += self.check_attribute(a, file, au.type)
 
         for s in au.statements:
             errors += self.check_element(s, file)
@@ -166,7 +201,7 @@ def check_dependency(self, d: Dependency, file: str) -> list[Error]:
         pass
 
     @abstractmethod
-    def check_attribute(self, a: Attribute, file: str) -> list[Error]:
+    def check_attribute(self, a: Attribute, file: str, au_type: None, parent_name: str = "") -> list[Error]:
         pass
 
     @abstractmethod
@@ -187,6 +222,9 @@ def check_comment(self, c: Comment, file: str) -> list[Error]:
 Error.agglomerate_errors()
 
 class SmellChecker(ABC):
+    def __init__(self) -> None:
+        self.code = None
+
     @abstractmethod
     def check(self, element, file: str) -> list[Error]:
         pass