463dff883b
- FastAPI backend: auth, findings, dashboard, attack paths, footprint, AI translator, risk calculator, PDF report generator - Next.js frontend: Vault dashboard, login, findings table, finding detail with AI coach, digital footprint, reports - PostgreSQL data model: tenants, users, assets, findings, risk scores, audit reports, attack paths - Docker Compose + Dockerfiles for all services - Demo seed data: Acme Corp with 6 findings and 90-day risk score history - AI Risk Translator (OpenAI/Anthropic) with plain-English business impact - Role-based access: executive / it_admin / trustos_admin - Scope-lock engine: authorization required before any assessment Stage 1-8 complete: Phase 1 Vault Audit product ready
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
"""
|
||
Risk score calculator service.
|
||
Score is 0–100, higher = safer (inverted from CVSS).
|
||
Scoring: Start at 100, deduct per open finding by severity.
|
||
"""
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, func
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
from app.db.session import AsyncSessionLocal
|
||
from app.models.models import Finding, RiskScore, FindingStatus, FindingCategory
|
||
|
||
|
||
SEVERITY_DEDUCTIONS = {
|
||
"critical": 12,
|
||
"high": 6,
|
||
"medium": 2,
|
||
"low": 0.5,
|
||
"info": 0,
|
||
}
|
||
|
||
CATEGORY_WEIGHTS = {
|
||
"external_exposure": 1.2,
|
||
"cloud_posture": 1.1,
|
||
"credential_exposure": 1.15,
|
||
"digital_footprint": 0.9,
|
||
"web_application": 1.1,
|
||
"network": 1.0,
|
||
"identity": 1.1,
|
||
"third_party": 0.8,
|
||
"compliance": 0.7,
|
||
"other": 0.6,
|
||
}
|
||
|
||
|
||
async def recalculate_risk_score(tenant_id: str, db: Optional[AsyncSession] = None) -> float:
|
||
"""Recalculate and persist the risk score for a tenant. Returns the new overall score."""
|
||
own_session = db is None
|
||
if own_session:
|
||
db = AsyncSessionLocal()
|
||
|
||
try:
|
||
# Fetch all open/in-progress findings
|
||
result = await db.execute(
|
||
select(Finding).where(
|
||
Finding.tenant_id == tenant_id,
|
||
Finding.status.in_([FindingStatus.open, FindingStatus.in_progress])
|
||
)
|
||
)
|
||
findings = result.scalars().all()
|
||
|
||
score = 100.0
|
||
category_scores: dict[str, float] = {}
|
||
severity_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
|
||
|
||
for f in findings:
|
||
sev = f.severity.value
|
||
cat = f.category.value
|
||
deduction = SEVERITY_DEDUCTIONS.get(sev, 0) * CATEGORY_WEIGHTS.get(cat, 1.0)
|
||
score -= deduction
|
||
if sev in severity_counts:
|
||
severity_counts[sev] += 1
|
||
|
||
score = max(0.0, min(100.0, round(score, 1)))
|
||
|
||
# Category sub-scores
|
||
category_map = {
|
||
"credential_exposure": "score_credential",
|
||
"cloud_posture": "score_cloud",
|
||
"external_exposure": "score_network",
|
||
"web_application": "score_web",
|
||
"identity": "score_identity",
|
||
"digital_footprint": "score_digital_footprint",
|
||
"third_party": "score_third_party",
|
||
}
|
||
cat_deductions: dict[str, float] = {}
|
||
for f in findings:
|
||
col = category_map.get(f.category.value)
|
||
if col:
|
||
cat_deductions[col] = cat_deductions.get(col, 0) + SEVERITY_DEDUCTIONS.get(f.severity.value, 0)
|
||
|
||
snapshot = RiskScore(
|
||
tenant_id=tenant_id,
|
||
score_date=datetime.utcnow(),
|
||
overall_score=score,
|
||
score_identity=max(0, 100 - cat_deductions.get("score_identity", 0)),
|
||
score_cloud=max(0, 100 - cat_deductions.get("score_cloud", 0)),
|
||
score_network=max(0, 100 - cat_deductions.get("score_network", 0)),
|
||
score_web=max(0, 100 - cat_deductions.get("score_web", 0)),
|
||
score_credential=max(0, 100 - cat_deductions.get("score_credential", 0)),
|
||
score_digital_footprint=max(0, 100 - cat_deductions.get("score_digital_footprint", 0)),
|
||
score_third_party=max(0, 100 - cat_deductions.get("score_third_party", 0)),
|
||
critical_count=severity_counts["critical"],
|
||
high_count=severity_counts["high"],
|
||
medium_count=severity_counts["medium"],
|
||
low_count=severity_counts["low"],
|
||
)
|
||
db.add(snapshot)
|
||
await db.commit()
|
||
return score
|
||
finally:
|
||
if own_session:
|
||
await db.close()
|