Application Security Expert

# tests/test_auth_security.py
import pytest
from app.auth import SecureAuth, InputValidator

class TestPasswordSecurity:
    """Security tests for password handling"""

    def test_rejects_weak_password(self):
        """Password must meet minimum requirements"""
        auth = SecureAuth()
        with pytest.raises(ValueError, match="at least 12 characters"):
            auth.hash_password("short")

    def test_password_hash_uses_argon2(self):
        """Must use Argon2id algorithm"""
        auth = SecureAuth()
        hashed = auth.hash_password("SecurePassword123!")
        assert hashed.startswith("$argon2id$")

    def test_different_salts_per_hash(self):
        """Each hash must have unique salt"""
        auth = SecureAuth()
        hash1 = auth.hash_password("TestPassword123!")
        hash2 = auth.hash_password("TestPassword123!")
        assert hash1 != hash2

class TestInputValidation:
    """Security tests for input validation"""

    def test_rejects_sql_injection_in_email(self):
        """Must reject SQL injection attempts"""
        assert not InputValidator.validate_email("admin'--@test.com")

    def test_rejects_xss_in_username(self):
        """Must reject XSS payloads"""
        assert not InputValidator.validate_username("<script>alert(1)</script>")

    def test_sanitizes_html_output(self):
        """Must escape HTML characters"""
        result = InputValidator.sanitize_html("<script>alert(1)</script>")
        assert "<script>" not in result
        assert "&lt;script&gt;" in result

# app/auth.py - Implement to pass tests
from argon2 import PasswordHasher

class SecureAuth:
    def __init__(self):
        self.ph = PasswordHasher(time_cost=3, memory_cost=65536)

    def hash_password(self, password: str) -> str:
        if len(password) < 12:
            raise ValueError("Password must be at least 12 characters")
        return self.ph.hash(password)

# Run security tests
pytest tests/test_auth_security.py -v

# Run SAST analysis
semgrep --config=auto app/

# Run secrets detection
gitleaks detect --source=. --verbose

# Run dependency check
pip-audit

# Good: Scan only changed files
def incremental_sast_scan(changed_files: list[str]) -> list:
    results = []
    for file_path in changed_files:
        if file_path.endswith(('.py', '.js', '.ts')):
            results.extend(run_semgrep(file_path))
    return results

# Bad: Full codebase scan on every commit
def full_scan():
    return run_semgrep(".")  # Slow for large codebases

# Good: Cache scan results with file hash
import hashlib
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_vulnerability_check(file_hash: str, rule_version: str):
    return run_security_scan(file_hash)

def scan_with_cache(file_path: str):
    content = Path(file_path).read_bytes()
    file_hash = hashlib.sha256(content).hexdigest()
    return cached_vulnerability_check(file_hash, RULE_VERSION)

# Bad: Re-scan unchanged files
def scan_without_cache(file_path: str):
    return run_security_scan(file_path)  # Redundant work

# Good: Parallel scanning with thread pool
from concurrent.futures import ThreadPoolExecutor

def parallel_security_scan(files: list[str], max_workers: int = 4):
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(scan_single_file, files))
    return [r for r in results if r]

# Bad: Sequential scanning
def sequential_scan(files: list[str]):
    results = []
    for f in files:
        results.append(scan_single_file(f))  # Slow
    return results

# Good: Focus on high-risk areas
HIGH_RISK_PATTERNS = ['auth', 'crypto', 'sql', 'exec', 'eval']

def targeted_audit(codebase_path: str):
    high_risk_files = []
    for pattern in HIGH_RISK_PATTERNS:
        high_risk_files.extend(grep_files(codebase_path, pattern))
    return deep_scan(set(high_risk_files))

# Bad: Equal depth for all files
def unfocused_audit(codebase_path: str):
    return deep_scan_all(codebase_path)  # Wastes resources

# Good: Set resource limits
import resource

def scan_with_limits(file_path: str):
    # Limit memory to 512MB
    resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024, -1))
    # Limit CPU time to 30 seconds
    resource.setrlimit(resource.RLIMIT_CPU, (30, 30))
    return run_analysis(file_path)

# Bad: Unbounded resource usage
def scan_unbounded(file_path: str):
    return run_analysis(file_path)  # Can exhaust system

# ✅ SECURE: Comprehensive input validation
from typing import Optional
import re
from html import escape
from urllib.parse import urlparse

class InputValidator:
    """Secure input validation following allowlist approach"""

    @staticmethod
    def validate_email(email: str) -> bool:
        """Validate email using strict regex"""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email)) and len(email) <= 254

    @staticmethod
    def validate_username(username: str) -> bool:
        """Validate username - alphanumeric only, 3-20 chars"""
        pattern = r'^[a-zA-Z0-9_]{3,20}$'
        return bool(re.match(pattern, username))

    @staticmethod
    def sanitize_html(user_input: str) -> str:
        """Escape HTML to prevent XSS"""
        return escape(user_input)

    @staticmethod
    def validate_url(url: str, allowed_schemes: list = ['https']) -> bool:
        """Validate URL and check scheme"""
        try:
            parsed = urlparse(url)
            return parsed.scheme in allowed_schemes and bool(parsed.netloc)
        except Exception:
            return False

    @staticmethod
    def validate_integer(value: str, min_val: int = None, max_val: int = None) -> Optional[int]:
        """Safely parse and validate integer"""
        try:
            num = int(value)
            if min_val is not None and num < min_val:
                return None
            if max_val is not None and num > max_val:
                return None
            return num
        except (ValueError, TypeError):
            return None

# ❌ DANGEROUS: String concatenation (SQLi vulnerable)
def get_user_vulnerable(username):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    cursor.execute(query)  # Vulnerable to: ' OR '1'='1

# ✅ SECURE: Parameterized queries (prepared statements)
def get_user_secure(username):
    query = "SELECT * FROM users WHERE username = ?"
    cursor.execute(query, (username,))

# ✅ SECURE: ORM with parameterized queries
from sqlalchemy import text

def get_user_orm(session, username):
    # SQLAlchemy automatically parameterizes
    user = session.query(User).filter(User.username == username).first()
    return user

# ✅ SECURE: Raw query with parameters
def search_users(session, search_term):
    query = text("SELECT * FROM users WHERE username LIKE :pattern")
    results = session.execute(query, {"pattern": f"%{search_term}%"})
    return results.fetchall()

// ❌ DANGEROUS: Direct HTML insertion
element.innerHTML = 'Hello ' + name;  // Vulnerable to XSS

// ✅ SECURE: Use textContent (no HTML parsing)
element.textContent = 'Hello ' + name;

// ✅ SECURE: DOMPurify for rich HTML
import DOMPurify from 'dompurify';
const clean = DOMPurify.sanitize(html, {
  ALLOWED_TAGS: ['b', 'i', 'em', 'strong', 'a', 'p'],
  ALLOWED_ATTR: ['href']
});

// ✅ SECURE: React/Vue automatically escape {variables}

# ✅ SECURE: Password hashing with Argon2id
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
import secrets

class SecureAuth:
    def __init__(self):
        self.ph = PasswordHasher(time_cost=3, memory_cost=65536, parallelism=4)

    def hash_password(self, password: str) -> str:
        if len(password) < 12:
            raise ValueError("Password must be at least 12 characters")
        return self.ph.hash(password)

    def verify_password(self, password: str, hash: str) -> bool:
        try:
            self.ph.verify(hash, password)
            return True
        except VerifyMismatchError:
            return False

    def generate_secure_token(self, bytes_length: int = 32) -> str:
        return secrets.token_urlsafe(bytes_length)

# ❌ NEVER: hashlib.md5(password.encode()).hexdigest()

# ✅ SECURE: JWT implementation
import jwt
from datetime import datetime, timedelta
import secrets

class JWTManager:
    def __init__(self, secret_key: str, algorithm: str = 'HS256'):
        self.secret_key = secret_key
        self.algorithm = algorithm

    def create_access_token(self, user_id: int, roles: list) -> str:
        now = datetime.utcnow()
        payload = {
            'sub': str(user_id), 'roles': roles, 'type': 'access',
            'iat': now, 'exp': now + timedelta(minutes=15),
            'jti': secrets.token_hex(16)
        }
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)

    def verify_token(self, token: str, expected_type: str = 'access'):
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm],
                options={'verify_exp': True, 'require': ['sub', 'exp', 'type', 'jti']})
            if payload.get('type') != expected_type:
                return None
            return payload
        except jwt.InvalidTokenError:
            return None