chore: restructure and rename python checkers

checkers/python/csrf-exempt.test.py

@@ -0,0 +1,14 @@
+from django.http import HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+
+# <expect-error>
+@csrf_exempt
+def my_view(request):
+    return HttpResponse('Hello world')
+
+import django
+
+# <expect-error>
+@django.views.decorators.csrf.csrf_exempt
+def my_view2(request):
+    return HttpResponse('Hello world')

checkers/python/csrf-exempt.yml

@@ -0,0 +1,32 @@
+language: py
+name: csrf-exempt
+message: Detected usage of csrf_exempt which indicates no CSRF
+category: security
+
+pattern: |
+  (decorated_definition
+    (decorator
+      (identifier) @csrf)
+  (#eq? @csrf "csrf_exempt")) @csrf-exempt
+
+
+  (decorated_definition
+    (decorator
+      (attribute
+        object: (attribute
+          object: (attribute
+            object: (attribute
+              object: (identifier) @django
+              attribute: (identifier) @views)
+            attribute: (identifier) @decorator)
+          attribute: (identifier) @csrf)
+        attribute: (identifier) @csrf_exempt))
+  (#eq? @django "django")
+  (#eq? @views "views")
+  (#eq? @decorator "decorators")
+  (#eq? @csrf "csrf")
+  (#eq? @csrf_exempt "csrf_exempt")) @csrf-exempt
+
+
+description: |
+  The decorator `csrf_exempt` disables CSRF protection, making routes vulnerable to attacks. Instead, define the function without this decorator to prevent unauthorized requests.

checkers/python/distributed-security-required-encryption.test.py

@@ -0,0 +1,108 @@
+# from https://github.com/apache/airflow/commit/80a3d6ac78c5c13abb8826b9dcbe0529f60fed81/
+
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import distributed
+import subprocess
+import warnings
+
+from airflow import configuration
+from airflow.executors.base_executor import BaseExecutor
+
+
+class DaskExecutor(BaseExecutor):
+    """
+    DaskExecutor submits tasks to a Dask Distributed cluster.
+    """
+    def __init__(self, cluster_address=None):
+        if cluster_address is None:
+            cluster_address = configuration.conf.get('dask', 'cluster_address')
+        if not cluster_address:
+            raise ValueError(
+                'Please provide a Dask cluster address in airflow.cfg')
+        self.cluster_address = cluster_address
+        # ssl / tls parameters
+        self.tls_ca = configuration.get('dask', 'tls_ca')
+        self.tls_key = configuration.get('dask', 'tls_key')
+        self.tls_cert = configuration.get('dask', 'tls_cert')
+        super(DaskExecutor, self).__init__(parallelism=0)
+
+    def start(self):
+        if self.tls_ca or self.tls_key or self.tls_cert:
+            from distributed.security import Security
+            # <expect-error>
+            security = Security(
+                tls_client_key=self.tls_key,
+                tls_client_cert=self.tls_cert,
+                tls_ca_file=self.tls_ca,
+                require_encryption=False,
+            )
+        else:
+            security = None
+
+        self.client = distributed.Client(self.cluster_address, security=security)
+        self.futures = {}
+
+    def execute_async(self, key, command, queue=None, executor_config=None):
+        if queue is not None:
+            warnings.warn(
+                'DaskExecutor does not support queues. '
+                'All tasks will be run in the same cluster'
+            )
+
+        def airflow_run():
+            return subprocess.check_call(command, shell=True, close_fds=True)
+
+        future = self.client.submit(airflow_run, pure=False)
+        self.futures[future] = key
+
+    def _process_future(self, future):
+        if future.done():
+            key = self.futures[future]
+            if future.exception():
+                self.log.error("Failed to execute task: %s", repr(future.exception()))
+                self.fail(key)
+            elif future.cancelled():
+                self.log.error("Failed to execute task")
+                self.fail(key)
+            else:
+                self.success(key)
+            self.futures.pop(future)
+
+    def sync(self):
+        # make a copy so futures can be popped during iteration
+        for future in self.futures.copy():
+            self._process_future(future)
+
+    def end(self):
+        for future in distributed.as_completed(self.futures.copy()):
+            self._process_future(future)
+
+    def terminate(self):
+        self.client.cancel(self.futures.keys())
+        self.end()
+
+    def create_security_context(self):
+        # <expect-error>
+        return distributed.security.Security(
+            tls_ca_file="ca.pem",
+            tls_cert="cert.pem",
+            require_encryption=False,
+        )

checkers/python/distributed-security-required-encryption.yml

@@ -0,0 +1,35 @@
+language: py
+name: distributed-security-required-encryption
+message: Detected security context for Dask with `require-encryption` keyword as False
+
+pattern: |
+  (call
+    function: (identifier) @security
+    arguments: (argument_list
+      (_)*
+      (keyword_argument
+        name: (identifier) @reqenc
+        value: (false))
+      (_)*)
+    (#eq? @reqenc "require_encryption")) @distributed-security-required-encryption
+
+  (call
+    function: (attribute
+      object: (attribute
+        object: (identifier) @dist
+        attribute: (identifier) @security_mod)
+      attribute: (identifier) @security_method)
+    arguments: (argument_list
+      (_)*
+      (keyword_argument
+        name: (identifier) @reqenc
+        value: (false))
+      (_)*)
+    (#eq? @dist "distributed")
+    (#eq? @security_mod "security")
+    (#eq? @security_method "Security")
+    (#eq? @reqenc "require_encryption")
+    ) @distributed-security-required-encryption
+
+description: |
+  Setting `require_encryption` as False in Dask security may weaken encryption. Set it to True to ensure data protection.

checkers/python/django-class-custom-extends.test.py

@@ -0,0 +1,135 @@
+from django.db.models import (
+    CharField, Expression, Field, FloatField, Lookup, TextField, Value,
+)
+from django.db.models.expressions import CombinedExpression, Func, Subquery
+from django.db.models.functions import Cast, Coalesce
+
+# <no-error>
+class Star(CharField):
+    pass
+
+# <expect-error>
+class MoreStar(Star):
+    pass
+
+# <expect-error>
+class Position(Func):
+    function = 'POSITION'
+    template = "%(function)s('%(substring)s' in %(expressions)s)"
+
+    # todoruleid: extends-custom-expression
+    def __init__(self, expression, substring):
+        # substring=substring is a SQL injection vulnerability!
+        super().__init__(expression, substring=substring)
+
+# <expect-error>
+class Aggregate(Func):
+    template = '%(function)s(%(distinct)s%(expressions)s)'
+    contains_aggregate = True
+    name = None
+    filter_template = '%s FILTER (WHERE %%(filter)s)'
+    window_compatible = True
+    allow_distinct = False
+
+    def __init__(self, *expressions, distinct=False, filter=None, **extra):
+        if distinct and not self.allow_distinct:
+            raise TypeError("%s does not allow distinct." % self.__class__.__name__)
+        self.distinct = distinct
+        self.filter = filter
+        super().__init__(*expressions, **extra)
+
+# <expect-error>
+class CastToInteger(Func):
+    """
+    A helper class for casting values to signed integer in database.
+    """
+    function = 'CAST'
+    template = '%(function)s(%(expressions)s as %(integer_type)s)'
+
+    def __init__(self, *expressions, **extra):
+        super().__init__(*expressions, **extra)
+        self.extra['integer_type'] = 'INTEGER'
+
+    def as_mysql(self, compiler, connection):
+        self.extra['integer_type'] = 'SIGNED'
+        return super().as_sql(compiler, connection)
+
+# <expect-error>
+class DateFormat(Func):
+    function = 'DATE_FORMAT'
+    template = '%(function)s(%(expressions)s, "%(format)s")'
+
+    def __init__(self, *expressions, **extra):
+        strf = extra.pop('format', None)
+        extra['format'] = strf.replace("%", "%%")
+        extra['output_field'] = CharField()
+        super(DateFormat, self).__init__(*expressions, **extra)
+
+# <expect-error>
+class StrFtime(Func):
+    function = 'strftime'
+    template = '%(function)s("%(format)s", %(expressions)s)'
+
+    def __init__(self, *expressions, **extra):
+        strf = extra.pop('format', None)
+        extra['format'] = strf.replace("%", "%%")
+        extra['output_field'] = CharField()
+        super(StrFtime, self).__init__(*expressions, **extra)
+
+# <expect-error>
+class RandomUUID(Func):
+    template = 'GEN_RANDOM_UUID()'
+    output_field = UUIDField()
+
+# <expect-error>
+class SearchHeadline(Func):
+    function = 'ts_headline'
+    template = '%(function)s(%(expressions)s%(options)s)'
+    output_field = TextField()
+
+    def __init__(
+        self, expression, query, *, config=None, start_sel=None, stop_sel=None,
+        max_words=None, min_words=None, short_word=None, highlight_all=None,
+        max_fragments=None, fragment_delimiter=None,
+    ):
+        if not hasattr(query, 'resolve_expression'):
+            query = SearchQuery(query)
+        options = {
+            'StartSel': start_sel,
+            'StopSel': stop_sel,
+            'MaxWords': max_words,
+            'MinWords': min_words,
+            'ShortWord': short_word,
+            'HighlightAll': highlight_all,
+            'MaxFragments': max_fragments,
+            'FragmentDelimiter': fragment_delimiter,
+        }
+        self.options = {
+            option: value
+            for option, value in options.items() if value is not None
+        }
+        expressions = (expression, query)
+        if config is not None:
+            config = SearchConfig.from_parameter(config)
+            expressions = (config,) + expressions
+        super().__init__(*expressions)
+
+# <expect-error>
+class _Str(Func):
+
+    """Casts a value to the database's text type."""
+
+    function = "CAST"
+    template = "%(function)s(%(expressions)s as %(db_type)s)"
+
+    def __init__(self, expression):
+        super().__init__(expression, output_field=models.TextField())
+
+    def as_sql(self, compiler, connection):
+        self.extra["db_type"] = self.output_field.db_type(connection)
+        return super().as_sql(compiler, connection)
+
+# <expect-error>
+class SQCount(Subquery):
+    template = "(SELECT count(*) FROM (%(subquery)s) _count)"
+    output_field = IntegerField()

checkers/python/django-class-custom-extends.yml

@@ -0,0 +1,17 @@
+language: py
+name: django-class-custom-extends
+message: Detected class extension with custom expression
+category: security
+severity: warning
+
+pattern: |
+  (class_definition
+    name: (identifier)
+    superclasses: (argument_list
+      (_)*
+      (identifier) @superclasses
+      (_)*)
+  (#match? @superclasses "^(django\\.db\\.models\\.(expressions\\.)?)?(Func|Expression|Value|DurationValue|RawSQL|Star|Random|Col|Ref|ExpressionList|ExpressionWrapper|When|Case|Window|WindowFrame|RowRange|ValueRange|Subquery|Exists)$")) @django-class-custom-extends
+
+description: |
+  xtending Django's expression classes can introduce security risks, such as SQL injection, if user input is not properly sanitized. Ensure that any user input is validated and does not directly enter custom expressions.

checkers/python/empty-aes-key.test.py

@@ -0,0 +1,11 @@
+from Crypto.Ciphers import AES
+
+def bad1():
+    # <expect-error>
+    cipher = AES.new("", AES.MODE_CFB, iv)
+    msg = iv + cipher.encrypt(b'Attack at dawn')
+
+def ok1(key):
+    # <no-error>
+    cipher = AES.new(key, AES.MODE_EAX, nonce=nonce)
+    plaintext = cipher.decrypt(ciphertext)

checkers/python/empty-aes-key.yml

@@ -0,0 +1,21 @@
+language: py
+name: empty-aes-key
+message: Found empty AES key
+category: security
+
+pattern: |
+  (call
+    function: (attribute
+      object: (identifier) @aes
+      attribute: (identifier) @new)
+    arguments: (argument_list
+      .
+      (string
+        (string_start)
+        (string_end))
+      (_)*)
+    (#eq? @aes "AES")
+    (#eq? @new "new")) @empty-aes-key
+
+description: |
+  Possible empty AES encryption key detected. An empty key weakens encryption and makes data vulnerable to attackers. Use a strong, non-empty key for security.