// === A01: Broken Access Control ===
// VULNERABLE: Direct object reference without authorization check
app.get('/api/users/:id/profile', async (req, res) => {
  const profile = await db.getUserProfile(req.params.id);
  res.json(profile); // Anyone can access any user's profile
});

// SECURE: Authorization check using middleware + ownership verification
const requireAuth = (req: Request, res: Response, next: NextFunction) => {
  const token = req.headers.authorization?.replace('Bearer ', '');
  if (!token) return res.status(401).json({ error: 'Authentication required' });
  try {
    req.user = jwt.verify(token, process.env.JWT_SECRET!) as UserClaims;
    next();
  } catch {
    return res.status(401).json({ error: 'Invalid token' });
  }
};

app.get('/api/users/:id/profile', requireAuth, async (req, res) => {
  const targetId = req.params.id;
  // Ownership check: users can only access their own profile
  // Admins can access any profile
  if (req.user.id !== targetId && !req.user.roles.includes('admin')) {
    return res.status(403).json({ error: 'Access denied' });
  }
  const profile = await db.getUserProfile(targetId);
  if (!profile) return res.status(404).json({ error: 'Not found' });
  res.json(profile);
});


// === A03: Injection ===
// VULNERABLE: SQL injection via string concatenation
app.get('/api/search', async (req, res) => {
  const query = req.query.q as string;
  // NEVER DO THIS — attacker sends: ' OR 1=1; DROP TABLE users; --
  const results = await db.raw(`SELECT * FROM products WHERE name LIKE '%${query}%'`);
  res.json(results);
});

// SECURE: Parameterized queries — the database driver handles escaping
app.get('/api/search', async (req, res) => {
  const query = req.query.q as string;
  if (!query || query.length > 200) {
    return res.status(400).json({ error: 'Invalid search query' });
  }
  // Parameterized: query is data, not code
  const results = await db('products')
    .where('name', 'ilike', `%${query}%`)
    .limit(50);
  res.json(results);
});


// === A07: Identification and Authentication Failures ===
// VULNERABLE: Timing attack on password comparison
function checkPassword(input: string, stored: string): boolean {
  return input === stored; // Short-circuits on first mismatch — leaks password length
}

// SECURE: Constant-time comparison + proper hashing
import { timingSafeEqual, scryptSync, randomBytes } from 'crypto';

function hashPassword(password: string): string {
  const salt = randomBytes(32).toString('hex');
  const hash = scryptSync(password, salt, 64).toString('hex');
  return `${salt}:${hash}`;
}

function verifyPassword(password: string, storedHash: string): boolean {
  const [salt, hash] = storedHash.split(':');
  const inputHash = scryptSync(password, salt, 64);
  const storedBuffer = Buffer.from(hash, 'hex');
  // Constant-time comparison — same duration regardless of where mismatch occurs
  return timingSafeEqual(inputHash, storedBuffer);
}


// === A08: Software and Data Integrity Failures ===
// VULNERABLE: Deserializing untrusted data
app.post('/api/import', (req, res) => {
  // NEVER deserialize untrusted input with eval or unsafe deserializers
  const data = JSON.parse(req.body.payload);
  // If using YAML: yaml.load() is unsafe — use yaml.safeLoad()
  // If using pickle (Python): NEVER unpickle untrusted data
  processImport(data);
});

// SECURE: Schema validation on all deserialized input
import { z } from 'zod';

const ImportSchema = z.object({
  items: z.array(z.object({
    name: z.string().max(200),
    quantity: z.number().int().positive().max(10000),
    category: z.enum(['electronics', 'clothing', 'food']),
  })).max(1000),
  metadata: z.object({
    source: z.string().max(100),
    timestamp: z.string().datetime(),
  }),
});

app.post('/api/import', (req, res) => {
  const parsed = ImportSchema.safeParse(req.body);
  if (!parsed.success) {
    return res.status(400).json({ error: 'Invalid input', details: parsed.error.issues });
  }
  // parsed.data is guaranteed to match the schema — type-safe and validated
  processImport(parsed.data);
});

#!/usr/bin/env python3
"""
Dependency security scanner integration for CI/CD pipelines.
Wraps multiple SCA tools and enforces organizational policy.
"""

import json
import subprocess
import sys
from dataclasses import dataclass
from enum import Enum
from pathlib import Path


class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@dataclass
class VulnFinding:
    package: str
    version: str
    severity: Severity
    cve: str
    fixed_version: str
    description: str
    exploitable: bool = False


class DependencyScanner:
    """Unified dependency scanning with policy enforcement."""

    # SLA: max days to remediate by severity
    REMEDIATION_SLA = {
        Severity.CRITICAL: 7,
        Severity.HIGH: 30,
        Severity.MEDIUM: 90,
        Severity.LOW: 180,
    }

    # Known false positives or accepted risks (with justification)
    SUPPRESSED = {
        "CVE-2023-XXXXX": "Not exploitable in our configuration — validated by AppSec team 2024-01-15",
    }

    @staticmethod
    def audit_json(command: list[str], project_path: Path) -> dict:
        """Exit 1 may mean findings; tool errors are never a clean scan."""
        result = subprocess.run(
            command, cwd=project_path, capture_output=True, text=True
        )
        try:
            audit = json.loads(result.stdout)
        except json.JSONDecodeError as exc:
            raise RuntimeError(f"{command[0]} did not produce valid JSON") from exc
        if result.returncode not in (0, 1) or not isinstance(audit, dict) or audit.get("error"):
            raise RuntimeError(f"{command[0]} failed (exit {result.returncode})")
        return audit

    def scan_npm(self, project_path: Path) -> list[VulnFinding]:
        """Scan Node.js dependencies using npm audit's current JSON report."""
        audit = self.audit_json(["npm", "audit", "--json", "--omit=dev"], project_path)
        vulnerabilities = audit.get("vulnerabilities")
        if not isinstance(vulnerabilities, dict):
            raise RuntimeError("npm audit report is missing vulnerabilities")
        findings = []
        for package, vuln in vulnerabilities.items():
            # via contains advisory objects OR names of indirect dependencies.
            via = vuln.get("via", [])
            advisories = [item for item in via if isinstance(item, dict)]
            fix = vuln.get("fixAvailable")
            severity = vuln.get("severity")
            if severity not in {item.value for item in Severity} | {"info"}:
                raise RuntimeError(f"npm audit returned unknown severity for {package}")
            findings.append(VulnFinding(
                package=package,
                version=vuln.get("range", "unknown"),
                severity=Severity.LOW if severity == "info" else Severity(severity),
                cve=advisories[0].get("url", "N/A") if advisories else "N/A",
                fixed_version=(fix.get("version", "N/A") if isinstance(fix, dict)
                               else "available" if fix is True else "N/A"),
                description="; ".join(item.get("title", "") for item in advisories)
                    or "Indirect dependency vulnerability: " + ", ".join(map(str, via)),
            ))
        return findings

    def scan_python(self, project_path: Path) -> list[VulnFinding]:
        """Audit requirements, or the active environment with the project installed."""
        command = ["pip-audit", "--format=json", "--desc"]
        if (project_path / "requirements.txt").exists():
            command.extend(["-r", "requirements.txt"])
        audit = self.audit_json(command, project_path)
        dependencies = audit.get("dependencies")
        if not isinstance(dependencies, list):
            raise RuntimeError("pip-audit report is missing dependencies")
        findings = []
        for dependency in dependencies:
            if dependency.get("skip_reason"):
                raise RuntimeError(f"pip-audit skipped {dependency['name']}")
            vulnerabilities = dependency.get("vulns")
            if not isinstance(vulnerabilities, list):
                raise RuntimeError("pip-audit dependency is missing vulns")
            for vuln in vulnerabilities:
                findings.append(VulnFinding(
                    package=dependency["name"],
                    version=dependency["version"],
                    severity=Severity.HIGH,  # Conservative local policy, not tool-provided severity
                    cve=vuln["id"],
                    fixed_version=", ".join(vuln.get("fix_versions", [])) or "N/A",
                    description=vuln.get("description", ""),
                ))
        return findings

    def enforce_policy(self, findings: list[VulnFinding]) -> tuple[bool, list[str]]:
        """
        Apply organizational policy to scan results.
        Returns (pass/fail, list of policy violations).
        """
        violations = []
        for f in findings:
            # Skip suppressed CVEs
            if f.cve in self.SUPPRESSED:
                continue

            # Critical and High with known fix = must block
            if f.severity in (Severity.CRITICAL, Severity.HIGH) and f.fixed_version != "N/A":
                violations.append(
                    f"BLOCKED: {f.package}@{f.version} has {f.severity.value} "
                    f"vulnerability {f.cve} — fix available: {f.fixed_version}"
                )

            # Critical without fix = warn but allow (with tracking)
            elif f.severity == Severity.CRITICAL and f.fixed_version == "N/A":
                violations.append(
                    f"WARNING: {f.package}@{f.version} has CRITICAL vulnerability "
                    f"{f.cve} with no fix available — track for remediation"
                )

        passed = not any("BLOCKED" in v for v in violations)
        return passed, violations


def main():
    scanner = DependencyScanner()
    project = Path(".")

    # Detect project type and scan. An unavailable scanner, malformed report,
    # unsupported report shape, or skipped dependency leaves coverage incomplete.
    findings = []
    try:
        if (project / "package.json").exists():
            findings.extend(scanner.scan_npm(project))
        if (project / "requirements.txt").exists() or (project / "pyproject.toml").exists():
            findings.extend(scanner.scan_python(project))
    except (OSError, RuntimeError, KeyError, TypeError, ValueError) as exc:
        print(f"SCAN INCOMPLETE: {exc}", file=sys.stderr)
        sys.exit(2)

    # Enforce policy
    passed, violations = scanner.enforce_policy(findings)

    for v in violations:
        print(v)

    print(f"\nTotal findings: {len(findings)}")
    print(f"Policy violations: {len(violations)}")
    print(f"Result: {'PASS' if passed else 'FAIL'}")

    sys.exit(0 if passed else 1)


if __name__ == "__main__":
    main()

# Threat Model: [Feature/System Name]

## System Overview
**Description**: [What this system does]
**Data Classification**: [Public / Internal / Confidential / Restricted]
**Compliance Scope**: [PCI-DSS / HIPAA / SOC 2 / None]

## Architecture Diagram
[Include or reference a data flow diagram showing components, trust boundaries, and data flows]

## Assets
| Asset | Classification | Location | Owner |
|-------|---------------|----------|-------|
| User credentials | Restricted | Auth service DB | Identity team |
| Payment data | Restricted (PCI) | Payment processor | Payments team |
| User profiles | Confidential | Main DB | Product team |

## Trust Boundaries
1. Internet → Load balancer (untrusted → semi-trusted)
2. Load balancer → API gateway (semi-trusted → trusted)
3. API gateway → Internal services (trusted → trusted)
4. Internal services → Database (trusted → restricted)

## STRIDE Analysis

### Spoofing (Authentication)
| Threat | Component | Risk | Mitigation |
|--------|-----------|------|------------|
| Stolen JWT used to impersonate user | API Gateway | High | Short-lived tokens (15min), refresh token rotation, token binding to IP range |
| API key leaked in client code | Mobile app | High | Use OAuth2 PKCE flow, never embed secrets in client apps |

### Tampering (Integrity)
| Threat | Component | Risk | Mitigation |
|--------|-----------|------|------------|
| Request body modified in transit | All APIs | Medium | TLS 1.3 enforced, HMAC signature on sensitive operations |
| Database records modified by attacker | Database | Critical | Parameterized queries, row-level security, audit logging |

### Repudiation (Audit)
| Threat | Component | Risk | Mitigation |
|--------|-----------|------|------------|
| User denies making a transaction | Payment service | High | Immutable audit log with timestamps, user action signatures |
| Admin denies changing permissions | Admin panel | Medium | Admin actions logged to append-only store with admin identity |

### Information Disclosure (Confidentiality)
| Threat | Component | Risk | Mitigation |
|--------|-----------|------|------------|
| Error messages expose stack traces | API responses | Medium | Generic error responses in production, detailed logging server-side only |
| Database dump via SQL injection | User search | Critical | Parameterized queries, WAF rules, input validation |

### Denial of Service (Availability)
| Threat | Component | Risk | Mitigation |
|--------|-----------|------|------------|
| API rate limit bypass | API Gateway | High | Per-user rate limiting, request size limits, pagination enforcement |
| ReDoS via crafted input | Input validation | Medium | Use RE2 (linear-time regex), input length limits |

### Elevation of Privilege (Authorization)
| Threat | Component | Risk | Mitigation |
|--------|-----------|------|------------|
| IDOR: user accesses other users' data | Profile API | Critical | Authorization check on every request, ownership verification |
| Mass assignment: user sets admin role | User update API | High | Explicit allowlist of updatable fields, never bind request body directly to model |

## Security Requirements (from this threat model)
1. [ ] Implement JWT token binding with 15-minute expiry
2. [ ] Add parameterized queries for all database operations
3. [ ] Enable audit logging for all state-changing operations
4. [ ] Implement per-user rate limiting (100 req/min default)
5. [ ] Add authorization middleware that verifies resource ownership
6. [ ] Strip sensitive fields from API error responses in production