Spaces:
Sleeping
Sleeping
dhruvkachhela
refactor(backend): use asyncio and thread pools for background tasks, optimize scanner AST utils and checks
1948258 | from __future__ import annotations | |
| import re | |
| import bisect | |
| from pathlib import Path | |
| from typing import Iterable, List, Optional, Sequence | |
| from .models import Confidence, FileRole, Finding, ScanDomain, Severity | |
| TEXT_EXTENSIONS = { | |
| ".py", ".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs", ".json", ".yml", ".yaml", ".tf", ".go", | |
| ".java", ".kt", ".rb", ".php", ".sh", ".bash", ".ps1", ".env", ".txt", ".md", ".html", ".htm", | |
| ".css", ".scss", ".less", ".dockerfile", ".lock", ".toml", ".ini", ".cfg", ".conf", ".sql", | |
| } | |
| def normalize_source_type(raw: Optional[str]) -> str: | |
| mapping = { | |
| "live_url": "url", | |
| "url": "url", | |
| "github": "github", | |
| "github_repo": "github", | |
| "repo": "github", | |
| "local_upload": "local", | |
| "local": "local", | |
| } | |
| return mapping.get((raw or "").lower(), "url") | |
| def infer_file_role(path: str) -> FileRole: | |
| lower = path.replace("\\", "/").lower() | |
| if "/test" in lower or lower.endswith((".test.ts", ".test.tsx", ".spec.ts", ".spec.tsx", ".spec.js", ".spec.jsx")): | |
| return FileRole.TEST | |
| if any(token in lower for token in ("fixture", "fixtures", "__mocks__", "mock", "sample", "example")): | |
| return FileRole.FIXTURE | |
| if any(token in lower for token in (".github/workflows", "/.github/", "github-actions", "circleci", "jenkins")): | |
| return FileRole.CI | |
| if any(token in lower for token in ("dockerfile", ".tf", ".tfvars", "k8s", "kubernetes", "helm", "chart", "compose")): | |
| return FileRole.INFRA | |
| if any(token in lower for token in ("config", "settings", "env", "infra", "deployment", "deploy")): | |
| return FileRole.CONFIG | |
| if any(token in lower for token in ("vendor", "node_modules", "dist", "build", "coverage")): | |
| return FileRole.VENDOR | |
| return FileRole.PRODUCTION | |
| def guess_language(path: str) -> Optional[str]: | |
| suffix = Path(path).suffix.lower() | |
| return { | |
| ".py": "python", | |
| ".ts": "typescript", | |
| ".tsx": "tsx", | |
| ".js": "javascript", | |
| ".jsx": "jsx", | |
| ".json": "json", | |
| ".yml": "yaml", | |
| ".yaml": "yaml", | |
| ".tf": "hcl", | |
| ".html": "html", | |
| ".htm": "html", | |
| ".sh": "bash", | |
| ".sql": "sql", | |
| }.get(suffix) | |
| def is_text_path(path: str) -> bool: | |
| suffix = Path(path).suffix.lower() | |
| return suffix in TEXT_EXTENSIONS or Path(path).name.lower() in {"dockerfile", "makefile", "procfile"} | |
| def normalize_path(path: str) -> str: | |
| return path.replace("\\", "/").lstrip("./") | |
| def should_ignore_path(path: str) -> bool: | |
| lower = normalize_path(path).lower() | |
| return any(token in lower for token in ("/.git/", "/node_modules/", "/dist/", "/build/", "/coverage/", "/vendor/")) | |
| def snippet_for_line(content: str, line_number: Optional[int], radius: int = 2) -> str: | |
| if not line_number or line_number < 1: | |
| return content[:500] | |
| lines = content.splitlines() | |
| start = max(0, line_number - radius - 1) | |
| end = min(len(lines), line_number + radius) | |
| return "\n".join(lines[start:end])[:1500] | |
| def make_finding( | |
| *, | |
| title: str, | |
| description: str, | |
| severity: Severity, | |
| domain: ScanDomain, | |
| file_path: Optional[str] = None, | |
| line_number: Optional[int] = None, | |
| check_id: Optional[str] = None, | |
| check_category: Optional[str] = None, | |
| policy_reference: Optional[str] = None, | |
| explanation: str = "", | |
| suggested_fix: str = "", | |
| confidence: Confidence = Confidence.MEDIUM, | |
| file_role: Optional[FileRole] = None, | |
| category: Optional[str] = None, | |
| secret_type: Optional[str] = None, | |
| raw_secret_value: Optional[str] = None, | |
| taint_path: Optional[object] = None, | |
| confirmed_runtime: bool = False, | |
| is_false_positive: bool = False, | |
| false_positive_reason: Optional[str] = None, | |
| ) -> Finding: | |
| return Finding( | |
| title=title, | |
| description=description, | |
| severity=severity, | |
| domain=domain, | |
| file_path=file_path, | |
| line_number=line_number, | |
| check_id=check_id, | |
| check_category=check_category, | |
| policy_reference=policy_reference, | |
| explanation=explanation or description, | |
| suggested_fix=suggested_fix, | |
| confidence=confidence, | |
| file_role=file_role or (infer_file_role(file_path) if file_path else None), | |
| category=category, | |
| secret_type=secret_type, | |
| raw_secret_value=raw_secret_value, | |
| taint_path=taint_path, | |
| confirmed_runtime=confirmed_runtime, | |
| is_false_positive=is_false_positive, | |
| false_positive_reason=false_positive_reason, | |
| ) | |
| def dedupe_findings(findings: Sequence[Finding]) -> List[Finding]: | |
| deduped: dict[tuple[str, str, Optional[int], Optional[str]], Finding] = {} | |
| for finding in findings: | |
| key = ( | |
| finding.title.lower().strip(), | |
| normalize_path(finding.file_path or ""), | |
| finding.line_number, | |
| (finding.check_id or "").strip() or None, | |
| ) | |
| existing = deduped.get(key) | |
| if existing is None: | |
| deduped[key] = finding | |
| continue | |
| if severity_rank(finding.severity) > severity_rank(existing.severity): | |
| deduped[key] = finding | |
| return list(deduped.values()) | |
| def severity_rank(severity: Severity | str) -> int: | |
| value = severity.value if isinstance(severity, Severity) else str(severity) | |
| order = {"critical": 5, "high": 4, "medium": 3, "low": 2, "info": 1} | |
| return order.get(value, 0) | |
| def score_from_findings(findings: Sequence[Finding]) -> int: | |
| score = 100 | |
| for finding in findings: | |
| if finding.is_false_positive: | |
| continue | |
| severity = finding.severity.value if isinstance(finding.severity, Severity) else str(finding.severity) | |
| weight = {"critical": 25, "high": 15, "medium": 8, "low": 3, "info": 0}.get(severity, 0) | |
| if finding.file_role in {FileRole.TEST, FileRole.FIXTURE, FileRole.VENDOR}: | |
| weight = max(0, weight - 4) | |
| score -= weight | |
| return max(0, min(100, score)) | |
| def is_ignored_for_validation(finding: Finding) -> bool: | |
| role = finding.file_role or (infer_file_role(finding.file_path) if finding.file_path else None) | |
| return role in {FileRole.TEST, FileRole.FIXTURE, FileRole.VENDOR} | |
| def split_lines(content: str) -> list[str]: | |
| return content.splitlines() or [content] | |
| def matches_any(patterns: Iterable[str], text: str) -> bool: | |
| return any(re.search(pattern, text, flags=re.IGNORECASE | re.MULTILINE) for pattern in patterns) | |
| 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 | |