scanner/layer1_framework_auth.py

from __future__ import annotations
import re
import os
import bisect
from pathlib import Path
from typing import List
from .models import Finding, Severity, Confidence, ScanDomain
from .ast_utils import ASTContext

class _LineCounter:
    def __init__(self, content: str):
        self.newlines = [i for i, c in enumerate(content) if c == '\n']
    def line_of(self, pos: int) -> int:
        return bisect.bisect_right(self.newlines, pos) + 1

_lc_cache: dict = {}

def _ln(content: str, pos: int) -> int:
    c_hash = id(content)
    if c_hash not in _lc_cache:
        _lc_cache[c_hash] = _LineCounter(content)
    return _lc_cache[c_hash].line_of(pos)

def _read(repo_path: str, rel_path: str) -> str | None:
    try:
        return Path(os.path.join(repo_path, rel_path)).read_text(encoding="utf-8", errors="replace")
    except Exception:
        return None


# Route-bearing directory segments — checked against Path.parts (platform-agnostic)
_ROUTE_DIR_PARTS = frozenset({
    "api", "routes", "controllers", "handler", "handlers",
    "pages",   # Next.js pages/api
    "app",     # Next.js app router
    "router",
})

# Filename stems that indicate a route/controller file
_ROUTE_FILE_STEMS = frozenset({
    "routes", "router", "controller", "handler",
    "index",  # pages/api/index.ts etc.
})

def _is_api_or_route_file(rel_path: str) -> bool:
    """
    Platform-agnostic route file detector.
    Uses Path.parts instead of substring matching to avoid
    leading-slash sensitivity on Windows vs Linux.
    """
    try:
        parts = Path(rel_path).parts
    except Exception:
        return False

    parts_lower = [p.lower() for p in parts]

    # Check: does any directory segment indicate a routing layer?
    for i, part in enumerate(parts_lower[:-1]):  # exclude the filename
        if part in _ROUTE_DIR_PARTS:
            if part == "pages":
                if i + 1 < len(parts_lower) and parts_lower[i + 1] == "api":
                    return True
            elif part == "app":
                if i + 1 < len(parts_lower) and parts_lower[i + 1] == "api":
                    return True
            else:
                return True

    # Check: filename stem indicates a route handler
    if parts_lower:
        stem = Path(parts_lower[-1]).stem
        if stem == "index" or any(kw in stem for kw in ("route", "router", "controller", "handler", "api")):
            return True

    return False



# Public endpoint path segments — these are intentionally unauthenticated.
# We keep scanning and keep findings, but downgrade confidence and add context label.
_PUBLIC_ENDPOINT_SEGMENTS = frozenset({
    "webhook", "webhooks", "signup", "sign-up", "login", "sign-in",
    "signin", "callback", "register", "auth", "oauth", "verify",
    "reset-password", "forgot-password", "health", "healthcheck",
})

def _is_public_endpoint(rel_path: str) -> bool:
    """Returns True if the file path suggests a public/unauthenticated endpoint."""
    parts = Path(rel_path).parts
    return any(p.lower() in _PUBLIC_ENDPOINT_SEGMENTS for p in parts)


# ---------------------------------------------------------------------------
# Check #14 — Clerk Auth: handler must call auth() and check userId
# ---------------------------------------------------------------------------

_CLERK_IMPORT_RE = re.compile(r"from\s+['\"]@clerk/nextjs['\"]|require\(['\"]@clerk/nextjs['\"]", re.IGNORECASE)
_CLERK_AUTH_CALL_RE = re.compile(r"\bauth\(\)|currentUser\(\)|getAuth\(", re.IGNORECASE)
_CLERK_USERID_CHECK_RE = re.compile(r"userId\s*[!=]=|if\s*\(\s*!?\s*userId|userId\s*\?\.|!userId", re.IGNORECASE)

def _check_clerk_auth(content: str, rel_path: str) -> List[Finding]:
    findings: List[Finding] = []
    if not _is_api_or_route_file(rel_path):
        return findings
    if not _CLERK_IMPORT_RE.search(content):
        return findings

    is_public = _is_public_endpoint(rel_path)
    context_suffix = (
        " [Context: Expected public endpoint — verify auth is intentionally absent]"
        if is_public else ""
    )

    ext = Path(rel_path).suffix
    ctx = ASTContext(content, ext)
    if not ctx.is_valid():
        return findings

    funcs = ctx.find_all_functions()
    for func in funcs:
        func_text = ctx.get_node_text(func)
        
        # We only care about exported functions or ones that look like handlers
        if not re.search(r"req|res|NextRequest|NextResponse|GET|POST|PUT|DELETE|PATCH", func_text):
            continue

        if not _CLERK_AUTH_CALL_RE.search(func_text):
            findings.append(Finding(
                title="Clerk Auth Import Without auth() Call",
                description=(
                    f"A handler in {rel_path} does not call auth() or currentUser(). "
                    f"The route is likely unprotected.{context_suffix}"
                ),
                severity=Severity.HIGH,
                confidence=Confidence.LOW if is_public else Confidence.HIGH,
                domain=ScanDomain.SAST,
                check_id="VS-FW-014a",
                check_category="auth",
                requires_llm_gate=True,
                policy_reference="Clerk Auth — Handler Protection",
                file_path=rel_path,
                line_number=func.start_point[0] + 1,
                suggested_fix="Add: const { userId } = auth(); if (!userId) return new NextResponse('Unauthorized', {status: 401});"
            ))
        elif not _CLERK_USERID_CHECK_RE.search(func_text):
            findings.append(Finding(
                title="Clerk auth() Called But userId Not Validated",
                description=(
                    f"A handler in {rel_path} calls auth() but never checks if userId is null. "
                    "auth() returns {{ userId: null }} for unauthenticated requests.{context_suffix}"
                ),
                severity=Severity.HIGH,
                confidence=Confidence.LOW if is_public else Confidence.MEDIUM,
                domain=ScanDomain.SAST,
                check_id="VS-FW-014b",
                check_category="auth",
                requires_llm_gate=True,
                policy_reference="Clerk Auth — userId Null Check",
                file_path=rel_path,
                line_number=func.start_point[0] + 1,
                suggested_fix="Add: if (!userId) return new NextResponse('Unauthorized', {status: 401});"
            ))

    return findings


# ---------------------------------------------------------------------------
# Check #15 — NextAuth: unprotected API routes in /api/ (no getServerSession)
# ---------------------------------------------------------------------------

_NEXTAUTH_IMPORT_RE = re.compile(
    r"from\s+['\"]next-auth['\"]|from\s+['\"]next-auth/react['\"]|"
    r"getServerSession|getSession\(",
    re.IGNORECASE,
)
_SESSION_CHECK_RE = re.compile(
    r"getServerSession|getSession\(|useSession|session\?\.user|session\.user",
    re.IGNORECASE,
)
_NEXTAUTH_CONFIG_RE = re.compile(r"\[\.\.\.nextauth\]|authOptions|NextAuth\(", re.IGNORECASE)

def _check_nextauth_session(content: str, rel_path: str) -> List[Finding]:
    findings: List[Finding] = []
    if not _is_api_or_route_file(rel_path):
        return findings
    if _NEXTAUTH_CONFIG_RE.search(content):
        return findings

    # Skip non-Next.js files (e.g. Supabase Edge Functions, Deno, Express, FastAPI)
    # NextAuth checks only apply to Next.js API routes/pages
    is_nextjs = bool(re.search(r"from\s+['\"]next['\"]|from\s+['\"]next-auth['\"]", content)) or "pages/api" in rel_path.replace("\\", "/").lower()
    is_deno_edge = rel_path.replace("\\", "/").lower().startswith("supabase/functions/") or "deno" in content.lower()[:200]
    if not is_nextjs or is_deno_edge:
        return findings

    has_db = re.compile(r"prisma\.|supabase\.|db\.\w+\.|mongoose\.", re.IGNORECASE)
    has_session = re.compile(
        r"getServerSession|getSession\(|useSession|session\?\.user|session\.user",
        re.IGNORECASE,
    )

    # ✅ NEW: if session is checked anywhere in the file, don't flag anything
    # This handles module-level middleware patterns and top-of-file guards
    if has_session.search(content):
        return findings

    ext = Path(rel_path).suffix
    ctx = ASTContext(content, ext)
    if not ctx.is_valid():
        return findings

    is_public = _is_public_endpoint(rel_path)
    context_suffix = " [Context: Expected public endpoint]" if is_public else ""

    funcs = ctx.find_all_functions()
    for func in funcs:
        func_text = ctx.get_node_text(func)

        if not re.search(r"req|res|NextRequest|NextResponse|GET|POST|PUT|DELETE|PATCH", func_text):
            continue

        if has_db.search(func_text):
            findings.append(Finding(
                title="NextAuth — API Route Accesses DB Without Session Verification",
                description=(
                    f"A handler in {rel_path} performs database operations but does not call "
                    f"getServerSession() within its scope to verify the caller is authenticated.{context_suffix}"
                ),
                severity=Severity.HIGH,
                confidence=Confidence.LOW if is_public else Confidence.HIGH,
                domain=ScanDomain.SAST,
                check_id="VS-FW-015",
                check_category="auth",
                requires_llm_gate=True,
                policy_reference="NextAuth — Session-Protected Routes",
                file_path=rel_path,
                line_number=func.start_point[0] + 1,
                suggested_fix="Add session verification: const session = await getServerSession(req, res, authOptions); if (!session) return res.status(401);"
            ))
            break

    return findings










# def _check_nextauth_session(content: str, rel_path: str) -> List[Finding]:
#     findings: List[Finding] = []
#     if not _is_api_or_route_file(rel_path):
#         return findings
#     if _NEXTAUTH_CONFIG_RE.search(content):
#         return findings  

#     has_db = re.compile(r"prisma\.|supabase\.|db\.\w+\.|mongoose\.", re.IGNORECASE)
#     has_session = re.compile(r"getServerSession|getSession\(|useSession|session\?\.user|session\.user", re.IGNORECASE)

#     ext = Path(rel_path).suffix
#     ctx = ASTContext(content, ext)
#     if not ctx.is_valid():
#         return findings

#     is_public = _is_public_endpoint(rel_path)
#     context_suffix = " [Context: Expected public endpoint]" if is_public else ""

#     funcs = ctx.find_all_functions()
#     for func in funcs:
#         func_text = ctx.get_node_text(func)
        
#         # Only care about handlers (e.g. ones with req/res or exported as GET/POST)
#         if not re.search(r"req|res|NextRequest|NextResponse|GET|POST|PUT|DELETE|PATCH", func_text):
#             continue

#         if has_db.search(func_text) and not has_session.search(func_text):
#             findings.append(Finding(
#                 title="NextAuth — API Route Accesses DB Without Session Verification",
#                 description=(
#                     f"A handler in {rel_path} performs database operations but does not call "
#                     "getServerSession() within its scope to verify the caller is authenticated.{context_suffix}"
#                 ),
#                 severity=Severity.HIGH,
#                 confidence=Confidence.LOW if is_public else Confidence.HIGH,
#                 domain=ScanDomain.SAST,
#                 check_id="VS-FW-015",
#                 check_category="auth",
#                 requires_llm_gate=True,
#                 policy_reference="NextAuth — Session-Protected Routes",
#                 file_path=rel_path,
#                 line_number=func.start_point[0] + 1,
#                 suggested_fix="Add session verification: const session = await getServerSession(req, res, authOptions); if (!session) return res.status(401);"
#             ))
#             # Don't flood with findings for every helper function in the file
#             break
            
#     return findings


# ---------------------------------------------------------------------------
# Check #16 — Firebase: Overly Permissive Security Rules
# ---------------------------------------------------------------------------

_FIREBASE_WEAK_RULE_RE = re.compile(
    r"allow\s+(?:read|write|read\s*,\s*write)\s*:\s*if\s+true\b",
    re.IGNORECASE,
)

def _check_firebase_rules(all_files: List[str], repo_path: str) -> List[Finding]:
    findings: List[Finding] = []

    firebase_rule_files = [
        f for f in all_files
        if Path(f).name.lower() in ("firestore.rules", "database.rules.json", "storage.rules")
        or f.replace("\\", "/").lower().endswith(".rules")
    ]

    for rel_path in firebase_rule_files:
        content = _read(repo_path, rel_path)
        if not content:
            continue
        for m in _FIREBASE_WEAK_RULE_RE.finditer(content):
            findings.append(Finding(
                title="Firebase — Overly Permissive Security Rule (allow if true)",
                description=(
                    f"Rule at line {_ln(content, m.start())} in {rel_path} permits "
                    "unconditional read/write access. Any unauthenticated user can "
                    "read or overwrite the entire database/storage path."
                ),
                severity=Severity.CRITICAL,
                confidence=Confidence.HIGH,
                domain=ScanDomain.CONFIG,
                check_id="VS-FW-016",
                check_category="auth",
                requires_llm_gate=True,
                policy_reference="Firebase Security Rules — Principle of Least Privilege",
                file_path=rel_path,
                line_number=_ln(content, m.start()),
                suggested_fix=(
                    "Replace 'if true' with an authentication check: "
                    "allow read, write: if request.auth != null && request.auth.uid == userId; "
                    "Never deploy with 'if true' outside of local development."
                ),
            ))
    return findings


# ---------------------------------------------------------------------------
# Check #17 — Auth0: JWT Handler Without Scope Validation
# ---------------------------------------------------------------------------

_AUTH0_JWT_RE = re.compile(
    r"from\s+['\"]express-oauth2-jwt-bearer['\"]|"
    r"from\s+['\"]auth0['\"]|"
    r"checkJwt\b|auth0\.verify\b|jwksRsa\b",
    re.IGNORECASE,
)
_SCOPE_CHECK_RE = re.compile(
    r"requiredScopes\(|checkScopes\(|req\.auth\.payload\.scope|"
    r"hasScope\(|verifyScopes\(|scope.*includes",
    re.IGNORECASE,
)

def _check_auth0_scopes(content: str, rel_path: str) -> List[Finding]:
    findings: List[Finding] = []
    if not _is_api_or_route_file(rel_path):
        return findings
    if not _AUTH0_JWT_RE.search(content):
        return findings
    if not _SCOPE_CHECK_RE.search(content):
        findings.append(Finding(
            title="Auth0 JWT — No OAuth Scope Validation",
            description=(
                f"{rel_path} uses Auth0 JWT verification but never validates OAuth scopes. "
                "Any valid Auth0 token (even one for a different audience or low-privilege "
                "scope) can access this endpoint."
            ),
            severity=Severity.MEDIUM,
            confidence=Confidence.MEDIUM,
            domain=ScanDomain.SAST,
            check_id="VS-FW-017",
            check_category="auth",
            requires_llm_gate=True,
            policy_reference="OAuth2 — Scope Enforcement",
            file_path=rel_path,
            line_number=1,
            suggested_fix=(
                "Add scope validation after JWT verification: "
                "const { requiredScopes } = require('express-oauth2-jwt-bearer'); "
                "router.get('/admin', checkJwt, requiredScopes('admin'), handler);"
            ),
        ))
    return findings


# ---------------------------------------------------------------------------
# Check #4.1 — Missing Auth in Server Actions
# ---------------------------------------------------------------------------

_SERVER_ACTION_RE = re.compile(
    r"export\s+(?:async\s+)?(?:function|const)\s+(\w+)\s*(?:=\s*(?:async\s*)?(?:\([^)]*\)|[a-zA-Z_]\w*)\s*=>|\()",
    re.IGNORECASE,
)

def _check_missing_auth_server_actions(content: str, rel_path: str) -> List[Finding]:
    findings: List[Finding] = []
    
    file_has_use_server = '"use server"' in content or "'use server'" in content
    
    db_write_re = re.compile(
        r"\.insert\(|\.update\(|\.delete\(|\.create\(|\.upsert\(|\.destroy\(|db\..*?\.(?:add|delete|save|update)",
        re.IGNORECASE,
    )
    auth_check_re = re.compile(
        r"getServerSession|auth\(\)|currentUser\(\)|getAuth\(|req\.user|session\?\.user|session\.user|requireAuth|ensureAuth|protect|guard",
        re.IGNORECASE,
    )

    ext = Path(rel_path).suffix
    ctx = ASTContext(content, ext)
    if not ctx.is_valid():
        return findings

    funcs = ctx.find_all_functions()
    for func in funcs:
        func_text = ctx.get_node_text(func)
        func_has_use_server = '"use server"' in func_text or "'use server'" in func_text
        
        is_exported = bool(re.search(r"\bexport\b", func_text))
        is_server_action = func_has_use_server or (file_has_use_server and is_exported)
        
        if is_server_action:
            if db_write_re.search(func_text) and not auth_check_re.search(func_text):
                findings.append(Finding( title="Missing Authentication in Next.js Server Action",
                        description=(
                            f"Server Action in '{rel_path}' performs database write operations "
                            "but does not verify the caller's session within the function scope. "
                            "Any client can invoke Server Actions directly, bypassing frontend UI blocks."
                        ),
                        severity=Severity.CRITICAL,
                        confidence=Confidence.HIGH,
                        domain=ScanDomain.SAST,
                        check_id="VS-FW-021",
                        check_category="auth",
                        policy_reference="Next.js Server Actions — Protection",
                        file_path=rel_path,
                        line_number=func.start_point[0] + 1,
                        suggested_fix="Add an authentication check inside the Server Action: const { userId } = auth(); if (!userId) throw new Error('Unauthorized');"))
    return findings


# ---------------------------------------------------------------------------
# Check #4.2 — Middleware Auth Bypass
# ---------------------------------------------------------------------------

def _check_middleware_auth_bypass(content: str, rel_path: str) -> List[Finding]:
    findings: List[Finding] = []
    if Path(rel_path).name.lower() != "middleware.ts":
        return findings
        
    if "matcher" in content:
        # Match either matcher: [...] or matcher: '...' / matcher: "..."
        m_match = re.search(r"matcher\s*:\s*(?:\[([^\]]+)\]|['\"`]([^'\"`]+)['\"`])", content, re.DOTALL)
        if m_match:
            matcher_val = m_match.group(1) or m_match.group(2) or ""
            is_excluded = False
            if "/api/" not in matcher_val and "api" not in matcher_val:
                is_excluded = True
            elif "(?!api" in matcher_val or "(?!.*api" in matcher_val:
                is_excluded = True

            if is_excluded:
                findings.append(Finding(
                    title="Next.js Middleware Auth Bypass for API Routes",
                    description=(
                        "Next.js middleware matcher does not include '/api/' paths. "
                        "This means all API routes bypass global middleware authentication checks."
                    ),
                    severity=Severity.HIGH,
                    confidence=Confidence.HIGH,
                    domain=ScanDomain.CONFIG,
                    check_id="VS-FW-022",
                    check_category="auth",
                    policy_reference="Next.js Middleware — Matcher Config",
                    file_path=rel_path,
                    line_number=1,
                    suggested_fix=(
                        "Include api routes in the middleware matcher config: "
                        "matcher: ['/dashboard/:path*', '/api/:path*']"
                      ),
                  ))
    return findings


# ---------------------------------------------------------------------------
# Public entry point
# ---------------------------------------------------------------------------

def run_framework_auth_checks(repo_path: str, production_files: List[str], all_files: List[str]) -> List[Finding]:
    """
    Runs all framework-specific auth checks (Clerk, NextAuth, Firebase, Auth0).
    Called from orchestrator.py after Layer 0 indexing.
    """
    findings: List[Finding] = []
    _lc_cache.clear()

    # Firebase rules may be in config files (not just production_files)
    findings.extend(_check_firebase_rules(all_files, repo_path))

    for rel_path in all_files:
        content = _read(repo_path, rel_path)
        if not content:
            continue
        findings.extend(_check_middleware_auth_bypass(content, rel_path))

    for rel_path in production_files:
        content = _read(repo_path, rel_path)
        if not content:
            continue
        findings.extend(_check_clerk_auth(content, rel_path))
        findings.extend(_check_nextauth_session(content, rel_path))
        findings.extend(_check_auth0_scopes(content, rel_path))
        findings.extend(_check_missing_auth_server_actions(content, rel_path))

    return findings