v2/skills/security_review.py

"""Security review chip + the audit findings as structured data.

AUDIT_FINDINGS is the single source of truth for the 13 confirmed
findings from the Nexus audit (2026-07-10, adversarially red-teamed,
43% survival rate). engine/beliefs.py seeds BDI beliefs from this list;
the discipline gate checks drafts against the same list. One place to
update when the campus team fixes a finding.
"""

import re
from dataclasses import dataclass, field

from kintsugi_core import (
    BaseSkillChip,
    EFEWeights,
    SkillCapability,
    SkillContext,
    SkillDomain,
    SkillRequest,
    SkillResponse,
)


@dataclass(frozen=True)
class AuditFinding:
    id: str
    severity: str          # CRITICAL | HIGH | MEDIUM | LOW
    title: str
    area: str              # auth | webhook | transaction | realtime | schema | moderation | cost
    file_hint: str
    advice: str
    patterns: tuple = ()   # regexes that indicate the same mistake recurring


AUDIT_FINDINGS = [
    AuditFinding(
        id="C1", severity="CRITICAL", area="schema",
        title="Lecture instructor_id type mismatch — all lectures broken",
        file_hint="server/src/services/lectureCapture.ts:48",
        advice=("socket.userId is a numeric string but lecture_sessions."
                "instructor_id is a UUID column. Use consistent ID types; "
                "never compare socket IDs to DB UUIDs with ===."),
        patterns=(r"socket\.userId\s*===", r"instructor_id"),
    ),
    AuditFinding(
        id="C2", severity="CRITICAL", area="webhook",
        title="Webhook signature bypass when env var is unset",
        file_hint="server/src/routes/webhook.ts:219",
        advice=("Never skip signature verification when a secret env var "
                "is missing — reject the request instead, or require "
                "NODE_ENV=development for any bypass."),
        patterns=(r"if\s*\(\s*!\s*secret\s*\)\s*return\s+true",
                  r"WEBHOOK_SECRET\s*\|\|\s*['\"]['\"]"),
    ),
    AuditFinding(
        id="H1", severity="HIGH", area="auth",
        title="JWT_SECRET falls back to empty string",
        file_hint="server/src/middleware/auth.ts:12",
        advice=("Startup must be fatal (process.exit(1)) when JWT_SECRET "
                "is empty or a placeholder. Never sign or verify with a "
                "defaulted secret."),
        patterns=(r"JWT_SECRET\s*(\?\?|\|\|)\s*['\"]",),
    ),
    AuditFinding(
        id="H2", severity="HIGH", area="realtime",
        title="connect_error triggers rapid reconnection loop",
        file_hint="client/src/stores/presenceStore.ts:578",
        advice=("All reconnect paths need exponential backoff. Only "
                "refresh tokens on 401/403, not on every connect_error."),
        patterns=(r"connect_error", r"refreshToken\(\)\s*.*connect\(\)"),
    ),
    AuditFinding(
        id="H3", severity="HIGH", area="transaction",
        title="Gem debit without transaction protection",
        file_hint="server/src/routes/store.ts:1108",
        advice=("debitGems/creditGems plus any dependent operation must "
                "share one DB transaction. A crash between them loses "
                "currency permanently."),
        patterns=(r"debitGems\(", r"creditGems\("),
    ),
    AuditFinding(
        id="H4", severity="HIGH", area="realtime",
        title="Trade system has no accept handler — feature incomplete",
        file_hint="server/src/services/socketHandlers/tradeHandlers.ts",
        advice=("trade:create-offer exists but accept/decline/cancel do "
                "not. Don't build on the trade flow assuming it completes."),
        patterns=(r"trade:(accept|create)-offer",),
    ),
    AuditFinding(
        id="M1", severity="MEDIUM", area="auth",
        title="JWT_SECRET prefix leaked into NPC Matrix password",
        file_hint="server/src/services/shopkeeperTools.ts:368",
        advice=("Never derive user-visible values from signing secrets. "
                "Use a separate secret or an HMAC derivation."),
        patterns=(r"JWT_SECRET\?*\.slice\(",),
    ),
    AuditFinding(
        id="M2", severity="MEDIUM", area="auth",
        title="Matrix admins receive isAdmin=true in client response",
        file_hint="server/src/routes/auth.ts:87",
        advice=("Client-facing admin flags must reflect server-enforced "
                "roles only; Matrix server admin is not campus admin."),
        patterns=(r"isMatrixServerAdmin", r"isAdmin\s*[:=]\s*true"),
    ),
    AuditFinding(
        id="M3", severity="MEDIUM", area="auth",
        title="No security headers (helmet/CSP/HSTS missing)",
        file_hint="server/src/index.ts",
        advice="Add helmet middleware with an appropriate CSP.",
        patterns=(r"app\.use\(helmet",),
    ),
    AuditFinding(
        id="M4", severity="MEDIUM", area="moderation",
        title="Faculty can kick/ban other faculty and admins",
        file_hint="server/src/services/socketHandlers/facultyPanelHandlers.ts:259",
        advice=("Moderation handlers must check the TARGET's role, not "
                "just the actor's — no acting on equal/higher privilege."),
        patterns=(r"(kick|ban|timeout).*(faculty|moderator)",),
    ),
    AuditFinding(
        id="M5", severity="MEDIUM", area="cost",
        title="No per-student message cap on agent conversations",
        file_hint="server/src/services/socketHandlers/agentHandlers.ts",
        advice=("Every new LLM-calling path needs a per-student daily cap "
                "or token budget."),
        patterns=(r"npc:message", r"callLLM\("),
    ),
    AuditFinding(
        id="L1", severity="LOW", area="schema",
        title="Foreign keys without ON DELETE break agent deletion",
        file_hint="migrations 193/223/303",
        advice=("New FKs referencing agents (or similar parents) need "
                "ON DELETE CASCADE or explicit dependent cleanup."),
        patterns=(r"REFERENCES\s+\w+\s*\([^)]*\)\s*(?!.*ON DELETE)",),
    ),
    AuditFinding(
        id="L2", severity="LOW", area="transaction",
        title="Agent job payments without transaction",
        file_hint="server/src/services/agentAutonomy.ts:1704",
        advice=("gem_balance and earnings updates must share one "
                "transaction."),
        patterns=(r"gem_balance", r"total_gems_earned"),
    ),
]

# Recurring anti-patterns the audit told the team to grep for.
RECURRING_PATTERNS = [
    ("env_fallback_disables_security",
     r"(SECRET|_KEY|TOKEN)\w*\s*(\?\?|\|\|)\s*['\"]",
     "Env var fallback that silently degrades security (audit pattern 4)."),
    ("parseint_no_nan_guard",
     r"parseInt\(req\.params",
     "parseInt on route params without a NaN guard (audit pattern 3)."),
    ("socket_on_without_off",
     r"socket\.on\(",
     "Socket listener registration — confirm matching socket.off cleanup "
     "(audit pattern 2)."),
]

AUTH_SIGNALS = [
    r"\bJWT_SECRET\b", r"\bverifyToken\b", r"\brequireAdmin\b",
    r"\brequireModerator\b", r"\bauthMiddleware\b", r"\bsession\s*cookie\b",
    r"\brejectIfIneligible\b", r"\bBearer\b", r"\brefresh[_ ]?token\b",
    r"\bwebhook\b", r"\bsignature\b",
]


def findings_relevant_to(text: str) -> list:
    """Findings whose area keywords or patterns appear in the text."""
    hits = []
    lower = text.lower()
    for f in AUDIT_FINDINGS:
        matched = any(re.search(p, text, re.IGNORECASE) for p in f.patterns)
        area_hit = f.area in lower
        if matched or area_hit:
            hits.append({"id": f.id, "severity": f.severity,
                         "title": f.title, "advice": f.advice,
                         "pattern_matched": matched})
    return hits


def recurring_pattern_hits(text: str) -> list:
    hits = []
    for name, pattern, message in RECURRING_PATTERNS:
        if re.search(pattern, text):
            hits.append({"pattern": name, "message": message})
    return hits


def touches_auth(text: str) -> bool:
    return any(re.search(p, text, re.IGNORECASE) for p in AUTH_SIGNALS)


class SecurityReviewChip(BaseSkillChip):
    """Reviews a request (and any code in it) against the audit's
    confirmed findings and the campus auth architecture. Optionally greps
    the live repo for the same patterns near files the request names."""

    name = "security_review"
    description = "Audit-findings and auth-impact review"
    version = "2.0.0"
    domain = SkillDomain.SECURITY
    efe_weights = EFEWeights(
        mission_alignment=0.15, stakeholder_benefit=0.30,
        resource_efficiency=0.10, transparency=0.30, equity=0.15,
    )
    capabilities = [SkillCapability.READ_DATA]

    def __init__(self, repo_files=None):
        super().__init__()
        self.repo_files = repo_files  # tools.file_tools.RepoFiles or None

    async def handle(self, request: SkillRequest,
                     context: SkillContext) -> SkillResponse:
        question = context.metadata.get("question", request.raw_input)
        session = context.metadata.get("session")

        auth = touches_auth(question)
        findings = findings_relevant_to(question)
        recurring = recurring_pattern_hits(question)

        repo_hits = []
        if self.repo_files is not None and auth:
            # Where does the repo actually enforce auth today?
            for probe in (r"verifyToken", r"requireAdmin"):
                repo_hits.extend(self.repo_files.search(probe)[:5])

        if session:
            for f in findings:
                session.record_evidence("audit_finding", f["id"], self.name)

        report = {
            "auth_touched": auth,
            "relevant_findings": findings,
            "recurring_pattern_hits": recurring,
            "repo_auth_sites": repo_hits,
        }
        summary = (
            f"auth_touched={auth}; {len(findings)} audit finding(s) relevant; "
            f"{len(recurring)} recurring pattern hit(s)"
        )
        # NOTE: data IS the artifact — DAGExecutor assigns the whole data
        # dict to this node's single output key ("security_report").
        return SkillResponse(
            content=summary, success=True,
            data=report,
            requires_consensus=auth,
            consensus_action="auth_change_review" if auth else None,
        )