dhruvkachhela
refactor(backend): use asyncio and thread pools for background tasks, optimize scanner AST utils and checks
1948258
Raw
History Blame Contribute Delete
6.77 kB
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