What “AI Reads Your Code” Actually Means (And Why SAST Can’t)

# SAST detection model for SQL injection:

# SAST looks for patterns like these:
patterns = [
    r'execute\\s*\\(\\s*[f"\\'].*%s',              # String formatting in execute()
    r'cursor\\.execute\\s*\\(\\s*f["\\']',           # f-string in execute()
    r'\\.raw\\s*\\(\\s*[f"\\']',                     # f-string in ORM raw()
    r'session\\.execute\\s*\\(\\s*text\\s*\\(',       # SQLAlchemy text() with string
]

# If any pattern matches → flag as potential SQL injection
# The SAST doesn't know:
#   - Whether the flagged code is actually reachable
#   - Whether the input that reaches the sink is actually user-controlled
#   - Whether there's sanitization between the source and sink
#   - Whether the pattern is a false positive in this specific context

# Source code:
def get_user_document(request, doc_id):
    user_id = request.user.id
    doc = Document.objects.get(id=doc_id)
    return doc.content

# Corresponding AST (simplified):
FunctionDef(
    name='get_user_document',
    args=['request', 'doc_id'],
    body=[
        Assign(
            target=Name('user_id'),
            value=Attribute(
                value=Attribute(
                    value=Name('request'),
                    attr='user'
                ),
                attr='id'
            )
        ),
        Assign(
            target=Name('doc'),
            value=Call(
                func=Attribute(
                    value=Attribute(
                        value=Name('Document'),
                        attr='objects'
                    ),
                    attr='get'
                ),
                args=[],
                kwargs={'id': Name('doc_id')}  # ← doc_id used here
            )
        ),
        Return(
            value=Attribute(
                value=Name('doc'),
                attr='content'
            )
        )
    ]
)

import ast

def find_orm_calls_without_user_filter(source_code, orm_model_name):
    """
    Find all ORM .get() calls that don't include a user ownership filter.
    This is the programmatic detection of missing authorization.
    """
    tree = ast.parse(source_code)

    vulnerable_calls = []

    for node in ast.walk(tree):
        # Find all attribute access chains (e.g., Document.objects.get())
        if isinstance(node, ast.Call):
            func = node.func

            # Check if this is a Model.objects.get() or Model.objects.filter() call
            if (isinstance(func, ast.Attribute) and
                func.attr in ('get', 'filter', 'exclude', 'first', 'last') and
                isinstance(func.value, ast.Attribute) and
                func.value.attr == 'objects'):

                model_name = func.value.value.id if isinstance(func.value.value, ast.Name) else None

                if model_name == orm_model_name:
                    # Check keywords for user/owner filter
                    has_user_filter = any(
                        kw.arg in ('user', 'owner', 'user_id', 'created_by', 'tenant')
                        for kw in node.keywords
                    )

                    if not has_user_filter:
                        vulnerable_calls.append({
                            'line': node.lineno,
                            'col': node.col_offset,
                            'model': model_name,
                            'method': func.attr,
                            'has_user_filter': False,
                            'finding': f'Missing user ownership filter in {model_name}.objects.{func.attr}()',
                            'recommendation': f'Add user={request.user} or owner={request.user} to filter'
                        })

    return vulnerable_calls

# Example usage:
source = open('views/documents.py').read()
findings = find_orm_calls_without_user_filter(source, 'Document')

for finding in findings:
    print(f"Line {finding['line']}: {finding['finding']}")
# Output:
# Line 47: Missing user ownership filter in Document.objects.get()
# Line 93: Missing user ownership filter in Document.objects.filter()
# Line 156: Missing user ownership filter in Document.objects.first()

import ast
from collections import defaultdict

class CallGraphBuilder(ast.NodeVisitor):
    """Build a call graph from Python source code"""

    def __init__(self):
        self.call_graph = defaultdict(set)  # function → set of functions it calls
        self.current_function = None
        self.function_locations = {}  # function name → (file, line)

    def visit_FunctionDef(self, node):
        """Record function definition"""
        previous_function = self.current_function
        self.current_function = node.name
        self.function_locations[node.name] = (self.current_file, node.lineno)

        # Visit the function body
        self.generic_visit(node)

        self.current_function = previous_function

    def visit_Call(self, node):
        """Record function calls"""
        if self.current_function:
            # Resolve the called function's name
            called_function = self.resolve_call_target(node)

            if called_function:
                self.call_graph[self.current_function].add(called_function)

        self.generic_visit(node)

    def resolve_call_target(self, call_node):
        """Extract the name of the function being called"""
        func = call_node.func

        if isinstance(func, ast.Name):
            return func.id
        elif isinstance(func, ast.Attribute):
            # method.call() → "method.call"
            if isinstance(func.value, ast.Name):
                return f"{func.value.id}.{func.attr}"
            elif isinstance(func.value, ast.Attribute):
                # chain.method.call()
                return f"{func.value.attr}.{func.attr}"

        return None

    def find_all_callers(self, target_function):
        """Find all functions that eventually call target_function"""
        callers = set()

        def dfs(func):
            for caller, callees in self.call_graph.items():
                if func in callees and caller not in callers:
                    callers.add(caller)
                    dfs(caller)  # Recursive — find callers of callers

        dfs(target_function)
        return callers

    def find_all_callees(self, source_function, depth=10):
        """Find all functions called by source_function (transitive)"""
        visited = set()

        def dfs(func, current_depth):
            if current_depth == 0 or func in visited:
                return
            visited.add(func)
            for callee in self.call_graph.get(func, set()):
                dfs(callee, current_depth - 1)

        dfs(source_function, depth)
        return visited - {source_function}

# Using the call graph for security analysis:

builder = CallGraphBuilder()
# ... (build graph from all source files)

# Question 1: Which HTTP endpoints eventually call cursor.execute()?
# (identifying all SQL execution paths from HTTP entry points)
endpoints = find_http_endpoints()  # returns view functions
sql_execution_paths = []

for endpoint in endpoints:
    callees = builder.find_all_callees(endpoint)
    if 'cursor.execute' in callees or 'cursor.executemany' in callees:
        sql_execution_paths.append({
            'endpoint': endpoint,
            'sql_functions': callees & {'cursor.execute', 'cursor.executemany'}
        })

# Question 2: Which functions call the dangerous subprocess.run?
# (finding command injection attack surface)
command_execution_callers = builder.find_all_callers('subprocess.run')
# Now check: do any of these callers receive user input?

# Question 3: Is the authentication check always executed before this admin function?
auth_callers = builder.find_all_callers('require_admin_auth')
admin_function_callers = builder.find_all_callers('admin_user_export')

# If admin_user_export has callers that don't call require_admin_auth → auth bypass
unprotected_paths = admin_function_callers - auth_callers

# Sources: Where user-controlled data enters the application
TAINT_SOURCES = {
    # Django
    'request.GET.get': 'user_input',
    'request.POST.get': 'user_input',
    'request.data.get': 'user_input',      # DRF
    'request.query_params.get': 'user_input',
    'request.json.get': 'user_input',      # Flask
    'request.form.get': 'user_input',      # Flask
    'request.args.get': 'user_input',      # Flask

    # Generic HTTP frameworks
    'request.body': 'user_input',
    'request.path': 'user_input',
    'request.headers.get': 'user_input',
    'request.cookies.get': 'user_input',

    # File uploads
    'request.FILES.get': 'user_input',
    'request.files.get': 'user_input',     # Flask
}

# Sinks: Where tainted data causes vulnerabilities when it arrives unsanitized
TAINT_SINKS = {
    # SQL injection sinks
    'cursor.execute': {'vuln_type': 'SQL Injection', 'severity': 'HIGH'},
    'cursor.executemany': {'vuln_type': 'SQL Injection', 'severity': 'HIGH'},
    'connection.execute': {'vuln_type': 'SQL Injection', 'severity': 'HIGH'},
    'Model.objects.raw': {'vuln_type': 'SQL Injection', 'severity': 'HIGH'},
    'session.execute': {'vuln_type': 'SQL Injection', 'severity': 'HIGH'},

    # Command injection sinks
    'os.system': {'vuln_type': 'Command Injection', 'severity': 'CRITICAL'},
    'subprocess.run': {'vuln_type': 'Command Injection', 'severity': 'CRITICAL'},
    'subprocess.Popen': {'vuln_type': 'Command Injection', 'severity': 'CRITICAL'},
    'subprocess.call': {'vuln_type': 'Command Injection', 'severity': 'CRITICAL'},
    'eval': {'vuln_type': 'Code Injection', 'severity': 'CRITICAL'},
    'exec': {'vuln_type': 'Code Injection', 'severity': 'CRITICAL'},

    # Path traversal sinks
    'open': {'vuln_type': 'Path Traversal', 'severity': 'HIGH'},
    'os.path.join': {'vuln_type': 'Path Traversal', 'severity': 'MEDIUM'},
    'os.listdir': {'vuln_type': 'Path Traversal', 'severity': 'MEDIUM'},
    'os.remove': {'vuln_type': 'Arbitrary File Deletion', 'severity': 'HIGH'},

    # SSRF sinks
    'requests.get': {'vuln_type': 'SSRF', 'severity': 'HIGH'},
    'requests.post': {'vuln_type': 'SSRF', 'severity': 'HIGH'},
    'urllib.request.urlopen': {'vuln_type': 'SSRF', 'severity': 'HIGH'},
    'httpx.get': {'vuln_type': 'SSRF', 'severity': 'HIGH'},

    # Template injection sinks
    'render_template_string': {'vuln_type': 'SSTI', 'severity': 'CRITICAL'},
    'Template': {'vuln_type': 'SSTI', 'severity': 'CRITICAL'},
    'jinja2.Template': {'vuln_type': 'SSTI', 'severity': 'CRITICAL'},

    # Deserialization sinks
    'pickle.loads': {'vuln_type': 'Unsafe Deserialization', 'severity': 'CRITICAL'},
    'pickle.load': {'vuln_type': 'Unsafe Deserialization', 'severity': 'CRITICAL'},
    'yaml.load': {'vuln_type': 'Unsafe Deserialization', 'severity': 'HIGH'},
    'marshal.loads': {'vuln_type': 'Unsafe Deserialization', 'severity': 'CRITICAL'},

    # XSS sinks (server-side rendering)
    'mark_safe': {'vuln_type': 'XSS', 'severity': 'HIGH'},
    'Markup': {'vuln_type': 'XSS', 'severity': 'HIGH'},
}

class TaintPropagationEngine:
    """
    Track tainted data through the application.
    Follows user input from source to sink through all transformations.
    """

    def __init__(self, ast_trees, call_graph):
        self.ast_trees = ast_trees
        self.call_graph = call_graph
        self.taint_facts = {}  # variable → taint_label
        self.findings = []

    def analyze_function(self, function_node, initial_taint=None):
        """
        Analyze a function for taint flow.
        initial_taint: dict of parameter_name → taint_label
        """
        taint_state = initial_taint or {}

        for stmt in ast.walk(function_node):

            # Assignment: propagate taint from right to left
            if isinstance(stmt, ast.Assign):
                for target in stmt.targets:
                    if self.is_tainted(stmt.value, taint_state):
                        taint_label = self.get_taint_label(stmt.value, taint_state)

                        if isinstance(target, ast.Name):
                            taint_state[target.id] = taint_label

            # Call: check if tainted data reaches a sink
            elif isinstance(stmt, ast.Call):
                sink_name = self.resolve_call_name(stmt)

                if sink_name in TAINT_SINKS:
                    # Check if any argument is tainted
                    for i, arg in enumerate(stmt.args):
                        if self.is_tainted(arg, taint_state):
                            self.report_finding(
                                sink=sink_name,
                                taint_label=self.get_taint_label(arg, taint_state),
                                sink_info=TAINT_SINKS[sink_name],
                                argument_index=i,
                                location=stmt
                            )

                    for kw in stmt.keywords:
                        if self.is_tainted(kw.value, taint_state):
                            self.report_finding(
                                sink=sink_name,
                                taint_label=self.get_taint_label(kw.value, taint_state),
                                sink_info=TAINT_SINKS[sink_name],
                                keyword=kw.arg,
                                location=stmt
                            )

                # Inter-procedural: propagate taint into called functions
                elif sink_name in self.call_graph:
                    tainted_args = {
                        i: self.get_taint_label(arg, taint_state)
                        for i, arg in enumerate(stmt.args)
                        if self.is_tainted(arg, taint_state)
                    }

                    if tainted_args:
                        # Recursively analyze the called function with tainted params
                        called_func = self.find_function(sink_name)
                        if called_func:
                            param_taint = {
                                called_func.args.args[i].arg: label
                                for i, label in tainted_args.items()
                                if i < len(called_func.args.args)
                            }
                            self.analyze_function(called_func, param_taint)

        return taint_state

    def is_tainted(self, node, taint_state):
        """Check if an AST node represents tainted data"""
        if isinstance(node, ast.Name):
            return node.id in taint_state

        elif isinstance(node, ast.Call):
            func_name = self.resolve_call_name(node)
            return func_name in TAINT_SOURCES

        elif isinstance(node, ast.JoinedStr):  # f-string
            # f-string is tainted if any of its values are tainted
            return any(
                self.is_tainted(value, taint_state)
                for value in node.values
                if isinstance(value, ast.FormattedValue)
            )

        elif isinstance(node, ast.BinOp):  # String concatenation
            return (self.is_tainted(node.left, taint_state) or
                    self.is_tainted(node.right, taint_state))

        elif isinstance(node, ast.Subscript):  # dict['key'] or list[i]
            return self.is_tainted(node.value, taint_state)

        elif isinstance(node, ast.Attribute):  # obj.attr
            return self.is_tainted(node.value, taint_state)

        return False

# views.py — HTTP entry point
class ProductSearchView(APIView):
    def get(self, request):
        search_term = request.GET.get('q', '')          # TAINT SOURCE
        category = request.GET.get('category', '')      # TAINT SOURCE
        sort_field = request.GET.get('sort_field', 'name')  # TAINT SOURCE

        results = ProductSearchService.search(
            query=search_term,
            category=category,
            sort=sort_field       # ← taint propagates into search()
        )
        return Response(results)

# services/product_search.py — taint travels here
class ProductSearchService:
    @staticmethod
    def search(query, category, sort='name'):
        # Applies basic text filtering (safe — uses ORM)
        queryset = Product.objects.filter(
            name__icontains=query,      # Safe — ORM parameterizes
            category=category           # Safe — ORM parameterizes
        )

        # Passes sort to the ordering method — taint travels further
        return ProductSearchService._apply_ordering(queryset, sort)

    @staticmethod
    def _apply_ordering(queryset, sort_field):
        # TAINT ARRIVES HERE
        # sort_field originated from request.GET — user controlled

        # SINK: raw SQL with f-string — SQL injection
        return queryset.extra(
            order_by=[f"products_product.{sort_field}"]  # ← VULNERABLE
        )

# Auth config analysis results:

auth_config = {
    'protected_paths': [
        {
            'pattern': '/api/public/**',
            'requirement': 'NONE (permitAll)',
            'is_public': True
        },
        {
            'pattern': '/api/v1/admin/**',
            'requirement': 'ROLE_ADMIN',
            'is_public': False
        },
        {
            'pattern': '/api/v1/users/**',
            'requirement': 'ROLE_USER',
            'is_public': False
        },
        {
            'pattern': '/api/**',  # anyRequest
            'requirement': 'authenticated',
            'is_public': False
        }
    ],

    'security_exclusions': [
        {
            'pattern': '/api/v2/**',
            'excluded_from': 'ALL security filters',
            'impact': 'CRITICAL — completely bypasses authentication',
            'finding': 'Any endpoint under /api/v2/ is accessible without credentials'
        },
        {
            'pattern': '/actuator/health',
            'excluded_from': 'ALL security filters',
            'impact': 'LOW — health endpoint, typically low sensitivity'
        }
    ]
}

# Cross-reference with discovered routes:
# From controller scan:
#   /api/v2/admin/users      → GET, POST — returns all users, creates users
#   /api/v2/data/export      → GET — exports complete dataset
#   /api/v2/internal/config  → GET — returns application configuration

# Result: Security exclusion of /api/v2/** makes ALL of these public
# Finding: Authentication bypass via webSecurityCustomizer exclusion
# CVSS: 9.8 (Critical) — unauthenticated access to admin, export, config endpoints

# Django settings.py — middleware configuration

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    # ← AuthenticationMiddleware is here
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    # Missing: Permission enforcement middleware
    # Authentication only → no authorization
]

# The AI reads every view in the application and checks for missing decorators

# views/admin.py — what the analysis finds:

class AdminUserExportView(View):
    # MISSING: @method_decorator(login_required, name='dispatch')
    # MISSING: @method_decorator(permission_required('admin.view_user'), name='dispatch')

    def get(self, request):
        # This view is completely public despite handling admin data
        users = User.objects.all().values(
            'id', 'email', 'first_name', 'last_name',
            'phone', 'address', 'date_joined'
        )
        return JsonResponse({'users': list(users)})

class UserProfileView(View):
    @method_decorator(login_required)  # ✓ Protected
    def dispatch(self, *args, **kwargs):
        return super().dispatch(*args, **kwargs)

    def get(self, request, user_id):
        # Protected by login_required but missing ownership check
        user = User.objects.get(id=user_id)  # ← IDOR
        return JsonResponse(model_to_dict(user))

# Middleware ordering analysis for Express.js apps:

def analyze_express_middleware_order(ast_tree):
    """
    Reconstruct the order in which middleware and routes are registered.
    Routes registered before a middleware don't go through it.
    """
    registration_order = []

    # Walk the AST looking for app.use() and app.get/post/etc() calls
    for node in ast.walk(ast_tree):
        if isinstance(node, ast.Call):
            if is_express_route_registration(node) or is_middleware_registration(node):
                registration_order.append({
                    'type': 'middleware' if is_middleware_registration(node) else 'route',
                    'path': extract_path(node),
                    'handler': extract_handler_name(node),
                    'line': node.lineno
                })

    # Find routes registered before authentication middleware
    auth_middleware_line = None
    for item in registration_order:
        if item['type'] == 'middleware' and is_auth_middleware(item['handler']):
            auth_middleware_line = item['line']
            break

    unprotected_routes = []
    if auth_middleware_line:
        for item in registration_order:
            if (item['type'] == 'route' and
                item['line'] < auth_middleware_line):
                unprotected_routes.append({
                    'path': item['path'],
                    'line': item['line'],
                    'finding': 'Route registered before authentication middleware'
                })

    return unprotected_routes

# Result:
# Unprotected route: GET /api/admin/users (line 8, before auth at line 21)
# Unprotected route: GET /api/admin/export (line 13, before auth at line 21)
# Finding: Authentication bypass via middleware ordering
# Impact: Two admin endpoints fully accessible without credentials

class ControlFlowAnalyzer:
    """
    Build a Control Flow Graph (CFG) and determine reachability.
    """

    def __init__(self):
        self.cfg = {}  # block_id → {successors, predecessors, statements}
        self.block_counter = 0

    def build_cfg(self, function_node):
        """Build CFG for a function"""
        entry_block = self.new_block()

        for stmt in function_node.body:
            if isinstance(stmt, ast.If):
                # Branch: create two paths, merge at convergence
                then_block = self.new_block()
                else_block = self.new_block()
                merge_block = self.new_block()

                self.add_edge(entry_block, then_block, condition=stmt.test)
                self.add_edge(entry_block, else_block, condition=f"NOT {stmt.test}")
                self.add_edge(then_block, merge_block)
                self.add_edge(else_block, merge_block)

            elif isinstance(stmt, ast.Try):
                # Exception handling: normal path and exception path
                try_block = self.new_block()
                except_block = self.new_block()

                self.add_edge(entry_block, try_block)
                self.add_edge(try_block, except_block, condition='exception')

    def is_reachable_from_http(self, function_name, target_line):
        """
        Determine if target_line is reachable from an HTTP endpoint
        without an intervening authentication check.
        """
        # Find all HTTP endpoints (Django views, Flask routes, Express handlers)
        http_endpoints = self.find_http_endpoints()

        for endpoint in http_endpoints:
            # BFS through call graph from endpoint to target function
            path = self.find_path(
                source=endpoint,
                target=function_name
            )

            if path:
                # Check if any step in the path includes an auth check
                auth_checks = self.find_auth_checks_in_path(path)

                if not auth_checks:
                    return {
                        'reachable': True,
                        'path': path,
                        'auth_protected': False,
                        'finding': f'{function_name} reachable from {endpoint} without auth'
                    }
                else:
                    return {
                        'reachable': True,
                        'path': path,
                        'auth_protected': True,
                        'auth_check_location': auth_checks
                    }

        return {'reachable': False}

SANITIZERS = {
    # SQL injection sanitizers
    'parameterize': 'sql_safe',
    'escape': 'sql_safe',
    'quote': 'sql_safe',

    # HTML/XSS sanitizers
    'escape_html': 'xss_safe',
    'bleach.clean': 'xss_safe',
    'html.escape': 'xss_safe',
    'markupsafe.escape': 'xss_safe',

    # Path traversal sanitizers
    'os.path.basename': 'path_safe',
    'werkzeug.utils.secure_filename': 'path_safe',

    # Command injection sanitizers
    'shlex.quote': 'cmd_safe',
    'pipes.quote': 'cmd_safe',
}

def check_sanitization_on_path(taint_path, sanitizers=SANITIZERS):
    """
    Check if user input is sanitized before reaching the sink.
    Returns None if unsafe, or the sanitizer if safe.
    """
    for step in taint_path:
        if isinstance(step, ast.Call):
            func_name = resolve_call_name(step)

            if func_name in sanitizers:
                sanitizer_label = sanitizers[func_name]

                # Check if the sanitizer covers the sink's vulnerability type
                if is_adequate_sanitizer(sanitizer_label, sink_vulnerability_type):
                    return {
                        'sanitized': True,
                        'sanitizer': func_name,
                        'label': sanitizer_label
                    }

    return None  # No adequate sanitization found

# This is why AI taint analysis has near-zero false positives:
# A flagged SQL injection only appears in the report if:
# 1. User input reaches the SQL sink (confirmed by taint propagation)
# 2. No adequate SQL sanitizer appears on the path (confirmed by sanitization check)
# 3. The code path is reachable from an HTTP endpoint (confirmed by CFG)
# All three conditions must be true simultaneously