安全 Risks of AI-Assisted Refactoring

# Original: constant-time comparison (secure but "ugly")
def verify_signature(provided: bytes, expected: bytes) -> bool:
    if len(provided) != len(expected):
        return False
    result = 0
    for a, b in zip(provided, expected):
        result |= a ^ b
    return result == 0
 
# AI-refactored: "simplified" but introduces timing side-channel
def verify_signature(provided: bytes, expected: bytes) -> bool:
    return provided == expected

# Original: uses hmac.compare_digest (constant-time)
import hmac
 
def verify_webhook(payload: bytes, signature: str, secret: bytes) -> bool:
    expected = hmac.new(secret, payload, "sha256").hexdigest()
    return hmac.compare_digest(expected, signature)
 
# AI-refactored: "simplified" string comparison
def verify_webhook(payload: bytes, signature: str, secret: bytes) -> bool:
    expected = hmac.new(secret, payload, "sha256").hexdigest()
    return expected == signature  # Timing-vulnerable

# Original: catch-all that prevents information disclosure
def authenticate(username: str, password: str) -> dict:
    try:
        user = db.get_user(username)
        if not user:
            raise AuthError("Invalid credentials")
        if not bcrypt.checkpw(password.encode(), user.password_hash):
            raise AuthError("Invalid credentials")
        return create_session(user)
    except AuthError:
        raise
    except Exception:
        # Intentionally generic to prevent information disclosure
        raise AuthError("Invalid credentials")
 
# AI-refactored: "better" error handling that leaks information
def authenticate(username: str, password: str) -> dict:
    user = db.get_user(username)
    if user is None:
        raise UserNotFoundError(f"User '{username}' not found")
 
    if not bcrypt.checkpw(password.encode(), user.password_hash):
        raise InvalidPasswordError("Incorrect password")
 
    return create_session(user)

# Original: atomic check-and-update (secure against race conditions)
def transfer_funds(from_id: int, to_id: int, amount: float, db):
    with db.transaction():
        result = db.execute(
            """
            UPDATE accounts SET balance = balance - %s
            WHERE id = %s AND balance >= %s
            RETURNING balance
            """,
            (amount, from_id, amount)
        )
        if not result:
            raise InsufficientFundsError()
        db.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_id)
        )
 
# AI-refactored: "cleaner" but introduces TOCTOU race condition
def transfer_funds(from_id: int, to_id: int, amount: float, db):
    sender = db.get_account(from_id)
 
    if sender.balance < amount:
        raise InsufficientFundsError(
            f"Balance {sender.balance} insufficient for transfer of {amount}"
        )
 
    # TOCTOU: balance could change between check above and update below
    sender.balance -= amount
    recipient = db.get_account(to_id)
    recipient.balance += amount
 
    db.save(sender)
    db.save(recipient)

# Original: 防禦-in-depth 授權
class DocumentController:
    def get_document(self, request, doc_id):
        # Layer 1: 認證 (verified by middleware)
        user = request.user
        if not user.is_authenticated:
            raise Unauthorized()
 
        # Layer 2: document exists
        doc = self.doc_repo.get(doc_id)
        if not doc:
            raise NotFound()
 
        # Layer 3: explicit 授權 check
        if doc.owner_id != user.id and not user.has_role("admin"):
            raise Forbidden()
 
        # Layer 4: field-level filtering
        return self._filter_sensitive_fields(doc, user)
 
    def _filter_sensitive_fields(self, doc, user):
        result = doc.to_dict()
        if not user.has_role("admin"):
            result.pop("internal_notes", None)
            result.pop("audit_trail", None)
        return result
 
# AI-refactored: "simplified" — removed "redundant" auth check
class DocumentController:
    def get_document(self, request, doc_id):
        # AI assumes middleware handles auth, removes explicit check
        doc = self.doc_repo.get(doc_id)
        if not doc:
            raise NotFound()
        # AI removed owner check 因為 "the API gateway handles 授權"
        return doc.to_dict()  # Also removed field filtering as "unnecessary"

# Original: Flask app with 安全 configuration
from flask import Flask
from flask_talisman import Talisman
 
app = Flask(__name__)
 
# 安全 headers
Talisman(app,
    content_security_policy={
        "default-src": "'self'",
        "script-src": "'self'",
        "style-src": "'self' 'unsafe-inline'",
    },
    force_https=True,
    strict_transport_security=True,
    strict_transport_security_max_age=31536000,
    session_cookie_secure=True,
    session_cookie_httponly=True,
    session_cookie_samesite="Lax",
)
 
app.config.update(
    SECRET_KEY=os.environ["SECRET_KEY"],
    SESSION_COOKIE_SECURE=True,
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SAMESITE="Lax",
    PERMANENT_SESSION_LIFETIME=timedelta(hours=1),
    MAX_CONTENT_LENGTH=16 * 1024 * 1024,
)
 
# AI-refactored: "cleaned up" — stripped "unnecessary" config
from flask import Flask
 
app = Flask(__name__)
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev-secret")

"""
Differential 測試 framework for detecting 安全 regressions
in AI-refactored code.
"""
 
import time
import statistics
from typing import Callable, Any
from dataclasses import dataclass
 
 
@dataclass
class DifferentialResult:
    test_name: str
    original_output: Any
    refactored_output: Any
    outputs_match: bool
    timing_difference_ns: float
    security_note: str
 
 
def differential_test(
    original_fn: Callable,
    refactored_fn: Callable,
    test_inputs: list[tuple],
    timing_samples: int = 100,
) -> list[DifferentialResult]:
    """
    Compare original and refactored functions for behavioral differences.
    Checks both 輸出 correctness and timing characteristics.
    """
    results = []
 
    for args in test_inputs:
        # Compare outputs
        try:
            orig_out = original_fn(*args)
        except Exception as e:
            orig_out = (type(e).__name__, str(e))
 
        try:
            ref_out = refactored_fn(*args)
        except Exception as e:
            ref_out = (type(e).__name__, str(e))
 
        # Compare timing (for constant-time operations)
        orig_times = []
        ref_times = []
        for _ in range(timing_samples):
            start = time.perf_counter_ns()
            try:
                original_fn(*args)
            except Exception:
                pass
            orig_times.append(time.perf_counter_ns() - start)
 
            start = time.perf_counter_ns()
            try:
                refactored_fn(*args)
            except Exception:
                pass
            ref_times.append(time.perf_counter_ns() - start)
 
        timing_diff = abs(
            statistics.median(orig_times) - statistics.median(ref_times)
        )
 
        security_notes = []
        if orig_out != ref_out:
            security_notes.append(
                f"輸出 changed: {orig_out!r} -> {ref_out!r}"
            )
 
        # Check if refactored version has higher timing variance
        # (may indicate loss of constant-time property)
        orig_cv = (statistics.stdev(orig_times) /
                   max(statistics.mean(orig_times), 1))
        ref_cv = (statistics.stdev(ref_times) /
                  max(statistics.mean(ref_times), 1))
        if ref_cv > orig_cv * 2:
            security_notes.append(
                "Timing variance increased significantly — "
                "possible loss of constant-time property"
            )
 
        results.append(DifferentialResult(
            test_name=f"args={args}",
            original_output=orig_out,
            refactored_output=ref_out,
            outputs_match=(orig_out == ref_out),
            timing_difference_ns=timing_diff,
            security_note="; ".join(security_notes) if security_notes else "OK",
        ))
 
    return results

"""
安全 property assertions that survive refactoring.
Place these in 測試 files that are run as part of CI/CD.
"""
 
import ast
import inspect
import dis
from pathlib import Path
 
 
def assert_constant_time_comparison(module, function_name: str):
    """
    Assert that a comparison function does not use Python's == operator
    on the compared values, which would not be constant-time.
    """
    source = inspect.getsource(getattr(module, function_name))
    tree = ast.parse(source)
 
    for node in ast.walk(tree):
        if isinstance(node, ast.Compare):
            for op in node.ops:
                if isinstance(op, (ast.Eq, ast.NotEq)):
                    # Check if the comparison involves the function's parameters
                    # 這是 a heuristic — may need refinement
                    raise AssertionError(
                        f"{function_name} uses == for comparison. "
                        f"Use hmac.compare_digest() or manual XOR loop "
                        f"for constant-time comparison. "
                        f"Line {node.lineno}"
                    )
 
 
def assert_generic_auth_errors(module, function_name: str):
    """
    Assert that an 認證 function does not raise specific
    error types that could enable username enumeration.
    """
    source = inspect.getsource(getattr(module, function_name))
    tree = ast.parse(source)
 
    forbidden_exceptions = {
        "UserNotFoundError", "InvalidPasswordError",
        "AccountDisabledError", "AccountLockedError",
    }
 
    for node in ast.walk(tree):
        if isinstance(node, ast.Raise) and node.exc:
            if isinstance(node.exc, ast.Call):
                if isinstance(node.exc.func, ast.Name):
                    if node.exc.func.id in forbidden_exceptions:
                        raise AssertionError(
                            f"{function_name} raises {node.exc.func.id}, "
                            f"which could enable username enumeration. "
                            f"Use generic AuthError instead. "
                            f"Line {node.lineno}"
                        )
 
 
def assert_transaction_wrapping(module, function_name: str):
    """
    Assert that a function that modifies multiple 資料庫 records
    does so within a transaction context manager.
    """
    source = inspect.getsource(getattr(module, function_name))
    tree = ast.parse(source)
 
    has_transaction = False
    for node in ast.walk(tree):
        if isinstance(node, ast.With):
            for item in node.items:
                if isinstance(item.context_expr, ast.Call):
                    func = item.context_expr.func
                    name = ""
                    if isinstance(func, ast.Attribute):
                        name = func.attr
                    elif isinstance(func, ast.Name):
                        name = func.id
                    if "transaction" in name.lower():
                        has_transaction = True
 
    if not has_transaction:
        raise AssertionError(
            f"{function_name} modifies 資料庫 records without "
            f"a transaction context manager. This could introduce "
            f"race conditions or partial update 漏洞."
        )

rules:
  - id: ai-refactor-timing-attack
    patterns:
      - pattern: |
          $SECRET == $USER_INPUT
      - metavariable-regex:
          metavariable: $SECRET
          regex: ".*(符元|secret|key|hash|signature|hmac|digest).*"
    message: >
      Direct comparison of 安全-sensitive value detected.
      This may have been introduced by AI refactoring.
      Use hmac.compare_digest() for constant-time comparison.
    severity: ERROR
    languages: [python]
 
  - id: ai-refactor-specific-auth-error
    patterns:
      - pattern: |
          raise UserNotFoundError(...)
      - pattern: |
          raise InvalidPasswordError(...)
    message: >
      Specific 認證 error type detected.
      AI refactoring may have introduced username enumeration.
      Use a generic AuthError with identical messages for all failure cases.
    severity: ERROR
    languages: [python]
 
  - id: ai-refactor-removed-csp
    patterns:
      - pattern-not: |
          content_security_policy = ...
      - pattern: |
          app = Flask(...)
    message: >
      Flask application without Content 安全 Policy.
      Verify CSP was not removed during AI refactoring.
    severity: WARNING
    languages: [python]
 
  - id: ai-refactor-default-secret
    pattern: |
      SECRET_KEY = os.environ.get("SECRET_KEY", "$DEFAULT")
    message: >
      SECRET_KEY has a default fallback value.
      AI refactoring may have added this. Use os.environ["SECRET_KEY"]
      to fail fast if the secret is not configured.
    severity: ERROR
    languages: [python]

# security_properties.py — checked into the repository
 
"""
SECURITY PROPERTIES — DO NOT REMOVE OR REFACTOR WITHOUT SECURITY REVIEW
 
This file documents 安全-critical properties of the codebase
that must be preserved during any refactoring, including AI-assisted.
These properties may not be obvious from the code alone.
 
1. auth.verify_password: Uses constant-time comparison (hmac.compare_digest).
   Reason: Prevents timing-based password guessing.
   測試: test_security/test_timing.py::test_password_comparison_constant_time
 
2. auth.authenticate: Returns identical errors for all failure modes.
   Reason: Prevents username enumeration.
   測試: test_security/test_auth.py::test_no_username_enumeration
 
3. payment.transfer: Executes within a 資料庫 transaction with SELECT FOR UPDATE.
   Reason: Prevents double-spend race conditions.
   測試: test_security/test_concurrency.py::test_concurrent_transfer
 
4. app.config: Must include CSP, HSTS, and secure cookie flags.
   Reason: 防禦-in-depth against XSS, MITM, and session hijacking.
   測試: test_security/test_headers.py::test_security_headers_present
"""

# Original: monolithic but secure — both checks in one atomic operation
class SecureAuthBackend(ModelBackend):
    def authenticate(self, request, username=None, password=None, totp=None):
        user = super().authenticate(request, username=username, password=password)
        if user is None:
            return None
        if not self._verify_totp(user, totp):
            return None  # MFA failure rejects even with correct password
        return user
 
# AI-refactored: modular but vulnerable — checks separated
class PasswordMiddleware:
    def process_request(self, request):
        # Authenticates via password, sets request.user
        pass
 
class MFAMiddleware:
    def process_request(self, request):
        # Checks MFA for request.user
        # But only runs for URLs in MFA_REQUIRED_PATHS
        pass

# Original custom logger with redaction
class PaymentLogger:
    REDACT_PATTERNS = [
        (r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', 'CARD-REDACTED'),
        (r'\b\d{3,4}\b(?=.*cvv)', 'CVV-REDACTED'),
    ]
 
    def info(self, message, **kwargs):
        sanitized = self._redact(str(kwargs))
        self._logger.info(message, extra={"data": sanitized})
 
    def _redact(self, text):
        for pattern, replacement in self.REDACT_PATTERNS:
            text = re.sub(pattern, replacement, text)
        return text
 
# AI-refactored: clean structured logging, no redaction
import structlog
logger = structlog.get_logger()
 
def process_payment(card_number, cvv, amount):
    logger.info("processing_payment",
                card=card_number,  # PCI data in plain text
                amount=amount)

# Original: distributed rate limiting via Redis Lua script
RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local window = tonumber(ARGV[1])
local limit = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
 
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)
 
if count < limit then
    redis.call('ZADD', key, now, now .. math.random())
    redis.call('EXPIRE', key, window)
    return 1
end
return 0
"""
 
# AI-refactored: "simplified" in-memory counter
from collections import defaultdict
import time
 
request_counts = defaultdict(list)  # Only works for single instance!
 
def check_rate_limit(client_id: str, limit: int = 100,
                     window: int = 60) -> bool:
    now = time.time()
    request_counts[client_id] = [
        t for t in request_counts[client_id] if t > now - window
    ]
    if len(request_counts[client_id]) >= limit:
        return False
    request_counts[client_id].append(now)
    return True

# Find potential AI-refactored commits
git log --all --oneline --grep="refactor" --grep="copilot" \
  --grep="ai" --grep="modernize" --grep="clean up" | head -20
 
# Find large structural changes
git log --all --numstat --since="6 months ago" | \
  awk '/^[0-9]/ {adds+=$1; dels+=$2} /^$/ {if(adds>100 && dels>100) print prev; adds=0; dels=0} {prev=$0}'

import subprocess
import re
 
 
def find_security_functions(repo_path: str) -> list[str]:
    """識別 安全-critical functions in the codebase."""
    security_indicators = [
        r"def (auth|login|verify|validate|encrypt|decrypt|hash|sign)",
        r"def .*(password|符元|secret|key|credential|session|權限)",
        r"def .*(sanitize|escape|filter|redact|rate.limit|csrf)",
    ]
    functions = []
    for pattern in security_indicators:
        result = subprocess.run(
            ["grep", "-rn", "-E", pattern, repo_path],
            capture_output=True, text=True
        )
        for line in result.stdout.splitlines():
            functions.append(line.strip())
    return functions
 
 
def diff_security_function(repo_path: str, function_name: str,
                           old_commit: str, new_commit: str) -> str:
    """Get the diff for a specific function between two commits."""
    result = subprocess.run(
        ["git", "-C", repo_path, "diff", old_commit, new_commit,
         "-U20", "--", "*.py"],
        capture_output=True, text=True
    )
    # Extract the relevant function diff
    # (simplified — real 實作 would use AST parsing)
    return result.stdout