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