398 lines
11 KiB
Python
398 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
from hashlib import sha256
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class Severity(StrEnum):
|
|
INFO = "info"
|
|
WARNING = "warning"
|
|
BLOCKER = "blocker"
|
|
|
|
|
|
class GateState(StrEnum):
|
|
PASS = "pass"
|
|
WARN = "warn"
|
|
BLOCK = "block"
|
|
|
|
|
|
class SourceOrigin(StrEnum):
|
|
LOCAL = "local"
|
|
GITEA = "gitea"
|
|
|
|
|
|
class ScanTrigger(StrEnum):
|
|
MANUAL = "manual"
|
|
SCHEDULED = "scheduled"
|
|
WEBHOOK = "webhook"
|
|
LOCAL = "local"
|
|
|
|
|
|
class ScanJobStatus(StrEnum):
|
|
QUEUED = "queued"
|
|
ACQUIRING_SOURCE = "acquiring_source"
|
|
SCANNING = "scanning"
|
|
ANALYZING = "analyzing"
|
|
PERSISTING = "persisting"
|
|
COMPLETED = "completed"
|
|
FAILED = "failed"
|
|
CANCELLED = "cancelled"
|
|
SUPERSEDED = "superseded"
|
|
|
|
|
|
class RepositoryState(StrEnum):
|
|
NEVER_SCANNED = "never_scanned"
|
|
CURRENT = "current"
|
|
STALE = "stale"
|
|
QUEUED = "queued"
|
|
SCANNING = "scanning"
|
|
FAILED = "failed"
|
|
UNAVAILABLE = "unavailable"
|
|
ARCHIVED = "archived"
|
|
|
|
|
|
class RepositoryPolicy(BaseModel):
|
|
auto_scan: bool = True
|
|
min_health_score: int = Field(default=0, ge=0, le=100)
|
|
max_blockers: int = Field(default=0, ge=0)
|
|
require_tests: bool = False
|
|
require_ci: bool = False
|
|
suppressions: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class RemediationSignal(BaseModel):
|
|
"""A compact, redacted finding summary safe for catalog-level triage."""
|
|
|
|
code: str
|
|
category: str
|
|
severity: Severity
|
|
message: str
|
|
count: int = Field(default=1, ge=1)
|
|
guidance: FindingGuidance | None = None
|
|
|
|
|
|
class ScanSource(BaseModel):
|
|
origin: SourceOrigin = SourceOrigin.LOCAL
|
|
provider: str | None = None
|
|
repository_id: str | None = None
|
|
repository_full_name: str | None = None
|
|
branch: str | None = None
|
|
requested_revision: str | None = None
|
|
actual_revision: str | None = None
|
|
trigger: ScanTrigger = ScanTrigger.MANUAL
|
|
|
|
|
|
class Repository(BaseModel):
|
|
id: str
|
|
provider: str
|
|
provider_repository_id: str
|
|
owner: str
|
|
name: str
|
|
full_name: str
|
|
description: str | None = None
|
|
html_url: str | None = None
|
|
clone_url: str | None = None
|
|
default_branch: str = "main"
|
|
private: bool = False
|
|
archived: bool = False
|
|
available: bool = True
|
|
latest_revision: str | None = None
|
|
latest_revision_at: datetime | None = None
|
|
last_synced_at: datetime | None = None
|
|
sync_error: str | None = None
|
|
state: RepositoryState = RepositoryState.NEVER_SCANNED
|
|
stale: bool = False
|
|
active_job_id: str | None = None
|
|
last_scan_id: str | None = None
|
|
last_scan_at: datetime | None = None
|
|
last_scan_health: int | None = None
|
|
last_scan_readiness: str | None = None
|
|
last_scan_revision: str | None = None
|
|
policy: RepositoryPolicy = Field(default_factory=RepositoryPolicy)
|
|
|
|
|
|
class RepositorySummary(Repository):
|
|
"""Catalog row intentionally contains no full scan report."""
|
|
|
|
last_scan_blockers: int = 0
|
|
last_scan_warnings: int = 0
|
|
last_scan_primary_signal: RemediationSignal | None = None
|
|
last_scan_health_delta: int | None = None
|
|
last_scan_added_findings: int = 0
|
|
last_scan_resolved_findings: int = 0
|
|
last_scan_priority: int = 0
|
|
last_scan_priority_rationale: str | None = None
|
|
|
|
|
|
class RepositoryDetail(RepositorySummary):
|
|
latest_scan: ScanReport | None = None
|
|
|
|
|
|
class ScanJob(BaseModel):
|
|
id: str
|
|
repository_id: str
|
|
status: ScanJobStatus
|
|
trigger: ScanTrigger
|
|
requested_revision: str | None = None
|
|
actual_revision: str | None = None
|
|
phase: str = "queued"
|
|
progress: int = Field(default=0, ge=0, le=100)
|
|
message: str | None = None
|
|
created_at: datetime
|
|
started_at: datetime | None = None
|
|
updated_at: datetime
|
|
finished_at: datetime | None = None
|
|
duration_ms: float | None = None
|
|
scan_id: str | None = None
|
|
error_code: str | None = None
|
|
error_detail: str | None = None
|
|
attempt: int = 0
|
|
deduplicated: bool = False
|
|
|
|
|
|
class RepositoryScanRequest(BaseModel):
|
|
force: bool = False
|
|
|
|
|
|
class RepositorySyncResult(BaseModel):
|
|
provider: str
|
|
discovered: int = 0
|
|
created: int = 0
|
|
updated: int = 0
|
|
unavailable: int = 0
|
|
errors: list[str] = Field(default_factory=list)
|
|
partial: bool = False
|
|
branch_failures: int = 0
|
|
listing_complete: bool = True
|
|
synced_at: datetime
|
|
|
|
|
|
class WebhookResponse(BaseModel):
|
|
accepted: bool
|
|
repository_id: str | None = None
|
|
job_id: str | None = None
|
|
message: str
|
|
|
|
|
|
class FindingGuidance(BaseModel):
|
|
"""Backend-owned, safe-to-display explanation of a deterministic rule."""
|
|
|
|
remediation_class: str
|
|
confidence: str
|
|
why: str
|
|
recommended_action: str
|
|
priority: int = Field(default=0, ge=0, le=1000)
|
|
priority_rationale: str
|
|
|
|
|
|
class Finding(BaseModel):
|
|
code: str
|
|
category: str
|
|
severity: Severity
|
|
message: str
|
|
path: str | None = None
|
|
line: int | None = None
|
|
deduction: int = Field(default=0, ge=0)
|
|
fingerprint: str | None = None
|
|
suppressed: bool = False
|
|
guidance: FindingGuidance | None = None
|
|
|
|
|
|
def finding_fingerprint(finding: Finding) -> str:
|
|
"""Build a stable semantic identity that ignores line movement and counters."""
|
|
normalized_message = re.sub(r"\b\d+(?:\.\d+)?\b", "#", finding.message.casefold())
|
|
payload = "\0".join(
|
|
(
|
|
finding.code,
|
|
finding.category,
|
|
finding.path or "",
|
|
normalized_message,
|
|
)
|
|
)
|
|
return sha256(payload.encode("utf-8")).hexdigest()[:32]
|
|
|
|
|
|
class RepositoryMetrics(BaseModel):
|
|
files: int = 0
|
|
source_files: int = 0
|
|
lines_of_code: int = 0
|
|
test_files: int = 0
|
|
todo: int = 0
|
|
fixme: int = 0
|
|
hack: int = 0
|
|
xxx: int = 0
|
|
dependency_manifests: int = 0
|
|
declared_dependencies: int = 0
|
|
large_files: int = 0
|
|
large_source_files: int = 0
|
|
|
|
|
|
class TestSignals(BaseModel):
|
|
frameworks: list[str] = Field(default_factory=list)
|
|
coverage_artifacts: list[str] = Field(default_factory=list)
|
|
ci_workflows: list[str] = Field(default_factory=list)
|
|
ci_validation_workflows: list[str] = Field(default_factory=list)
|
|
ci_build_workflows: list[str] = Field(default_factory=list)
|
|
ci_deploy_only_workflows: list[str] = Field(default_factory=list)
|
|
coverage_evidence: list[str] = Field(default_factory=list)
|
|
test_source_ratio: float | None = None
|
|
|
|
|
|
class GitSignals(BaseModel):
|
|
available: bool = False
|
|
commit_count: int | None = None
|
|
contributor_count: int | None = None
|
|
latest_commit_at: datetime | None = None
|
|
tracked_files: int | None = None
|
|
renamed_files: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class DependencySignals(BaseModel):
|
|
manifests: list[str] = Field(default_factory=list)
|
|
lockfiles: list[str] = Field(default_factory=list)
|
|
declared_dependencies: int = 0
|
|
unbounded_dependencies: int = 0
|
|
unpinned_dependencies: int = 0
|
|
duplicate_dependencies: list[str] = Field(default_factory=list)
|
|
included_requirements: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ArchitectureNode(BaseModel):
|
|
id: str
|
|
label: str
|
|
path: str
|
|
language: str
|
|
inbound: int = 0
|
|
outbound: int = 0
|
|
visible_inbound: int = 0
|
|
visible_outbound: int = 0
|
|
hotspot_score: int = Field(default=0, ge=0, le=100)
|
|
|
|
|
|
class ArchitectureEdge(BaseModel):
|
|
source: str
|
|
target: str
|
|
kind: str = "import"
|
|
|
|
|
|
class ArchitectureGraph(BaseModel):
|
|
nodes: list[ArchitectureNode] = Field(default_factory=list)
|
|
edges: list[ArchitectureEdge] = Field(default_factory=list)
|
|
clusters: dict[str, int] = Field(default_factory=dict)
|
|
truncated: bool = False
|
|
total_nodes: int = 0
|
|
total_edges: int = 0
|
|
|
|
|
|
class Hotspot(BaseModel):
|
|
path: str
|
|
language: str
|
|
lines_of_code: int
|
|
debt_markers: int = 0
|
|
churn: int = 0
|
|
inbound: int = 0
|
|
outbound: int = 0
|
|
score: int = Field(ge=0, le=100)
|
|
reasons: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class GateCheck(BaseModel):
|
|
key: str
|
|
label: str
|
|
state: GateState
|
|
detail: str
|
|
|
|
|
|
class ReleaseGate(BaseModel):
|
|
status: str
|
|
checks: list[GateCheck]
|
|
|
|
|
|
class ScanLimits(BaseModel):
|
|
max_files: int = Field(default=10_000, ge=1, le=100_000)
|
|
max_text_bytes_per_file: int = Field(default=2_000_000, ge=1_024, le=10_000_000)
|
|
max_total_text_bytes: int = Field(default=50_000_000, ge=1_000_000, le=500_000_000)
|
|
max_findings: int = Field(default=2_000, ge=10, le=20_000)
|
|
max_secret_findings: int = Field(default=20, ge=1, le=500)
|
|
|
|
|
|
class ScanCoverage(BaseModel):
|
|
limits: ScanLimits
|
|
discovered_files: int = 0
|
|
scanned_files: int = 0
|
|
skipped_files_by_budget: int = 0
|
|
skipped_files_by_size: int = 0
|
|
skipped_files_by_text_budget: int = 0
|
|
text_bytes_scanned: int = 0
|
|
truncated: bool = False
|
|
notes: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ScanReport(BaseModel):
|
|
id: str
|
|
repository_path: str
|
|
repository_name: str
|
|
source: ScanSource = Field(default_factory=ScanSource)
|
|
policy: RepositoryPolicy = Field(default_factory=RepositoryPolicy)
|
|
scanned_at: datetime
|
|
duration_ms: float
|
|
metrics: RepositoryMetrics
|
|
languages: dict[str, int]
|
|
category_scores: dict[str, int]
|
|
health_score: int
|
|
release_readiness: str
|
|
findings: list[Finding]
|
|
tests: TestSignals = Field(default_factory=TestSignals)
|
|
git: GitSignals = Field(default_factory=GitSignals)
|
|
dependencies: DependencySignals = Field(default_factory=DependencySignals)
|
|
architecture: ArchitectureGraph = Field(default_factory=ArchitectureGraph)
|
|
hotspots: list[Hotspot] = Field(default_factory=list)
|
|
release_gate: ReleaseGate | None = None
|
|
coverage: ScanCoverage | None = None
|
|
compacted: bool = False
|
|
|
|
|
|
class ScanSummary(BaseModel):
|
|
id: str
|
|
repository_id: str
|
|
scanned_at: datetime
|
|
branch: str | None = None
|
|
requested_revision: str | None = None
|
|
actual_revision: str | None = None
|
|
trigger: ScanTrigger = ScanTrigger.MANUAL
|
|
health_score: int
|
|
release_readiness: str
|
|
findings: int
|
|
duration_ms: float
|
|
|
|
|
|
class ScanRequest(BaseModel):
|
|
path: str = Field(min_length=1)
|
|
|
|
|
|
class ScanComparison(BaseModel):
|
|
current_scan_id: str
|
|
previous_scan_id: str | None = None
|
|
health_score_delta: int | None = None
|
|
category_score_deltas: dict[str, int] = Field(default_factory=dict)
|
|
metric_deltas: dict[str, int] = Field(default_factory=dict)
|
|
added_findings: list[Finding] = Field(default_factory=list)
|
|
resolved_findings: list[Finding] = Field(default_factory=list)
|
|
|
|
|
|
class TrendPoint(BaseModel):
|
|
scan_id: str
|
|
scanned_at: datetime
|
|
health_score: int
|
|
findings: int
|
|
blockers: int
|
|
warnings: int
|
|
debt_markers: int
|
|
|
|
|
|
RepositoryDetail.model_rebuild()
|