Initial implementation of transformation

Author: andrecsilva

src/core_codemods/docs/sonar_python_use-secure-protocols.md

@@ -0,0 +1,24 @@
+Communication using clear-text protocols allows an attacker to sniff or tamper with the transported data.
+
+This codemod will replace any detected clear text protocol with their cryptographic enabled version.
+
+Our changes look like the following:
+```diff
+- url = "http://example.com"
++ url = "https://example.com"
+
+- ftp_con = ftplib.FTP("ftp.example.com")
++ ftp_con = ftplib.FTP_TLS("ftp.example.com")
++ smtp_context = ssl.create_default_context()
++ smtp_context.verify_mode = ssl.CERT_REQUIRED
++ smtp_context.check_hostname = True
+smtp_con = smtplib.SMTP("smtp.example.com", port=587)
++ smtp.starttls(context=smtp_context)
+
+
++ smtp_context_1 = ssl.create_default_context()
++ smtp_context_1.verify_mode = ssl.CERT_REQUIRED
++ smtp_context_1.check_hostname = True
+- smtp_con_2 = smtplib.SMTP("smtp.gmail.com")
++ smtp_con_2 = smtplib.SMTP_SSL("smtp.gmail.com", context=smtp_context_1)
+```

src/core_codemods/sonar/api.py

@@ -10,6 +10,25 @@
 
 
 class SonarCodemod(SASTCodemod):
+
+    def __init__(
+        self,
+        *,
+        metadata: Metadata,
+        transformer: BaseTransformerPipeline,
+        default_extensions: list[str] | None = None,
+        requested_rules: list[str] | None = None,
+        provider: str | None = None,
+    ):
+        super().__init__(
+            metadata=metadata,
+            detector=SonarDetector(),
+            transformer=transformer,
+            default_extensions=default_extensions,
+            requested_rules=requested_rules,
+            provider=provider,
+        )
+
     @property
     def origin(self):
         return "sonar"
@@ -38,7 +57,6 @@ def from_core_codemod_with_multiple_rules(
                 ),
             ),
             transformer=transformer if transformer else other.transformer,
-            detector=SonarDetector(),
             default_extensions=other.default_extensions,
             requested_rules=[tr.id for tr in rules],
         )

src/core_codemods/sonar/sonar_use_secure_protocols.py

@@ -1,9 +1,15 @@
+import libcst as cst
+from libcst.codemod import CodemodContext
+
 from codemodder.codemods.base_codemod import Metadata, ReviewGuidance, ToolRule
 from codemodder.codemods.libcst_transformer import (
     LibcstResultTransformer,
     LibcstTransformerPipeline,
 )
+from codemodder.codemods.utils_mixin import NameAndAncestorResolutionMixin
 from codemodder.codetf import Reference
+from codemodder.file_context import FileContext
+from codemodder.result import Result
 from core_codemods.sonar.api import SonarCodemod
 
 rules = [
@@ -15,10 +21,154 @@
 ]
 
 
-class SonarUseSecureProtocolsTransformer(LibcstResultTransformer):
+class SonarUseSecureProtocolsTransformer(
+    LibcstResultTransformer, NameAndAncestorResolutionMixin
+):
     change_description = "Modified URLs or calls to use secure protocols"
 
+    def __init__(
+        self,
+        context: CodemodContext,
+        results: list[Result] | None,
+        file_context: FileContext,
+        _transformer: bool = False,
+    ):
+        self.nodes_memory_with_context_name: dict[cst.CSTNode, str] = {}
+        super().__init__(context, results, file_context, _transformer)
+
+    def _match_and_handle_statement(
+        self, possible_smtp_call, original_node_statement, updated_node_statement
+    ):
+        maybe_name = self.find_base_name(possible_smtp_call)
+        match possible_smtp_call:
+            case cst.Call() if maybe_name == "smtplib.SMTP":
+                # get the stored context_name or create a new one:
+                if possible_smtp_call in self.nodes_memory_with_context_name:
+                    context_name = self.nodes_memory_with_context_name[
+                        possible_smtp_call
+                    ]
+                else:
+                    context_name = self.generate_available_name(
+                        original_node_statement, ["smtp_context"]
+                    )
+
+                new_statements = []
+                new_statements.append(
+                    cst.parse_statement(
+                        f"{context_name} = ssl.create_default_context()"
+                    )
+                )
+                new_statements.append(
+                    cst.parse_statement(
+                        f"{context_name}.verify_mode = ssl.CERT_REQUIRED"
+                    )
+                )
+                new_statements.append(
+                    cst.parse_statement(f"{context_name}.check_hostname = True")
+                )
+                new_statements.append(updated_node_statement)
+                # don't append this if we changed the call to SSL version
+                if possible_smtp_call in self.nodes_memory_with_context_name:
+                    self.nodes_memory_with_context_name.pop(possible_smtp_call)
+                else:
+                    new_statements.append(
+                        cst.parse_statement(f"smtplib.starttls(context={context_name})")
+                    )
+                self.add_needed_import("smtplib")
+                self.add_needed_import("ssl")
+                self.report_change(possible_smtp_call)
+                return cst.FlattenSentinel(new_statements)
+        return updated_node_statement
+
+    def leave_SimpleStatementLine(self, original_node, updated_node):
+        match original_node.body:
+            # match the first statement that is either selected or is an assignment whose value is selected
+            case [cst.Assign() as a, *_] if self.node_is_selected(a.value):
+
+                return self._match_and_handle_statement(
+                    a.value, original_node, updated_node
+                )
+            case [s, *_] if self.node_is_selected(s):
+                return self._match_and_handle_statement(s, original_node, updated_node)
+        return updated_node
+
     def leave_Call(self, original_node, updated_node):
+        if self.node_is_selected(original_node):
+            match self.find_base_name(original_node):
+                case "ftplib.FTP":
+                    new_func = cst.parse_expression("ftplib.FTP_TLS")
+                    self.report_change(original_node)
+                    self.add_needed_import("ftplib")
+                    return updated_node.with_changes(func=new_func)
+                # Just using ssl.create_default_context() may not be enough for older python versions
+                # See https://stackoverflow.com/questions/33857698/sending-email-from-python-using-starttls
+                case "smtplib.SMTP":
+                    # port is the second positional, check that
+                    maybe_port_value = (
+                        original_node.args[1]
+                        if len(original_node.args) >= 2
+                        and original_node.args[1].keyword is None
+                        else None
+                    )
+                    # find port keyword, if any
+                    maybe_port_value = maybe_port_value or next(
+                        iter(
+                            [
+                                a
+                                for a in original_node.args
+                                if a.keyword and a.keyword.value == "port"
+                            ]
+                        ),
+                        None,
+                    )
+                    maybe_port_value = (
+                        maybe_port_value.value if maybe_port_value else None
+                    )
+                    match maybe_port_value:
+                        case None:
+                            self._change_to_smtp_ssl(original_node, updated_node)
+                        case cst.Integer() if maybe_port_value == "0":
+                            self._change_to_smtp_ssl(original_node, updated_node)
+
+        return updated_node
+
+    def _change_to_smtp_ssl(self, original_node, updated_node):
+        # remember this node so we don't add the starttls
+        new_func = cst.parse_expression("smtplib.SMTP_SSL")
+
+        context_name = self.generate_available_name(original_node, ["smtp_context"])
+        self.nodes_memory_with_context_name[original_node] = context_name
+
+        new_args = [
+            *original_node.args,
+            cst.Arg(
+                keyword=cst.Name("context"),
+                value=cst.Name(context_name),
+            ),
+        ]
+        return updated_node.with_changes(func=new_func, args=new_args)
+
+    def leave_SimpleString(
+        self, original_node: cst.SimpleString, updated_node: cst.SimpleString
+    ) -> cst.BaseExpression:
+        if self.node_is_selected(original_node):
+            match original_node.raw_value:
+                case original_node.raw_value as s if s.startswith("http"):
+                    self.report_change(original_node)
+                    return updated_node.with_changes(
+                        value=original_node.prefix
+                        + original_node.quote
+                        + s.replace("http", "https", 1)
+                        + original_node.quote
+                    )
+                case original_node.raw_value as s if s.startswith("ftp"):
+                    self.report_change(original_node)
+                    return updated_node.with_changes(
+                        value=original_node.prefix
+                        + original_node.quote
+                        + s.replace("ftp", "sftp", 1)
+                        + original_node.quote
+                    )
         return updated_node