diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..027affb --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,28 @@ +name: CI + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - run: python -m pip install --upgrade pip + - run: python -m pip install -e '.[dev]' + - run: ruff check . + - run: coverage run -m unittest discover -v + - run: coverage report + - run: python -m build + diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c143198 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +__pycache__/ +*.py[cod] +.coverage +.pytest_cache/ +.ruff_cache/ +.venv/ +build/ +dist/ +*.egg-info/ +report*.json + diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..54a8f76 --- /dev/null +++ b/LICENSE @@ -0,0 +1,22 @@ +MIT License + +Copyright (c) 2026 Alan Nogueira + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..8676d22 --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,6 @@ +include LICENSE +include README.md +include SECURITY.md +recursive-include docs *.md +recursive-include examples *.json +recursive-include tests *.py diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..7cd3775 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,34 @@ +# Segurança + +## Modelo operacional + +A versão `0.1.0` processa exportações JSON locais. Ela não solicita credenciais cloud e não chama +APIs da AWS, Azure ou Google Cloud. + +Use somente exportações obtidas com autorização. Achados podem conter identificadores de contas, +assinaturas, projetos, recursos, topologia e vulnerabilidades ainda não corrigidas. + +## Proteção dos relatórios + +- identificadores de achados e recursos são pseudonimizados; +- títulos e descrições originais não são gravados; +- relatórios são publicados atomicamente; +- em sistemas POSIX, o arquivo recebe modo `0600`; +- um segredo HMAC pode ser fornecido em `CLOUD_AUDIT_HMAC_SECRET`; +- o relatório continua confidencial mesmo com pseudonimização. + +Não publique exportações reais, relatórios ou segredos em commits, issues ou pull requests. + +## Limites + +- entrada máxima de 20 MiB; +- links simbólicos são rejeitados; +- não há execução de remediação; +- não há autenticação ou coleta automática nesta versão; +- o score não substitui análise de risco, pentest ou resposta a incidentes. + +## Comunicação responsável + +Não abra issue pública contendo dados do ambiente ou detalhes de uma vulnerabilidade explorável. +Use um canal privado previamente combinado com o mantenedor. + diff --git a/docs/SCORING.md b/docs/SCORING.md new file mode 100644 index 0000000..6db0aaa --- /dev/null +++ b/docs/SCORING.md @@ -0,0 +1,56 @@ +# Metodologia de score + +Versão: `cloud-risk-1.0.0`. + +O score é uma regra de priorização criada por este projeto. Ele não corresponde ao AWS Security +Hub score, Microsoft Secure Score ou Google Cloud Security Command Center score. Também não +conclui conformidade, comprometimento ou aceitabilidade do risco. + +## Score por achado + +| Fator | Pontos | +|---|---:| +| Severidade informativa | 0 | +| Severidade baixa | 10 | +| Severidade média | 25 | +| Severidade alta | 45 | +| Severidade crítica | 70 | +| Severidade desconhecida | 15 | +| Prova de conceito disponível | +5 | +| Exploit funcional | +10 | +| Exploração conhecida | +20 | +| Exposição pública explícita | +15 | +| Contexto privilegiado explícito | +10 | +| Dados sensíveis explícitos | +10 | +| Mais de 90 dias | +5 | +| Mais de 180 dias | +10 | +| Controle compensatório explícito | -10 | + +O resultado é limitado ao intervalo de 0 a 100. Achados resolvidos, inativos, saudáveis, +arquivados ou não aplicáveis recebem zero. + +## Faixas + +- 0: informativo; +- 1 a 19: baixo; +- 20 a 39: médio; +- 40 a 69: alto; +- 70 a 100: crítico. + +## Score do ambiente + +Somente achados ativos com pontuação positiva entram no cálculo: + +```text +environment_score = 50% maior risco + + 35% média dos 10 maiores riscos + + 15% proporção de achados críticos +``` + +O desenho evita que uma grande quantidade de alertas baixos esconda um achado crítico. + +## Confiança + +`confidence_score` mede completude da evidência, não risco. Fonte reconhecida, severidade, estado, +identificador do recurso, identificador do achado, timestamp e sinais contextuais contribuem para +a confiança. A confiança nunca é somada ao risco. diff --git a/examples/aws-securityhub.json b/examples/aws-securityhub.json new file mode 100644 index 0000000..e382591 --- /dev/null +++ b/examples/aws-securityhub.json @@ -0,0 +1,22 @@ +{ + "Findings": [ + { + "SchemaVersion": "2018-10-08", + "Id": "synthetic-aws-finding-001", + "Title": "Synthetic internet exposed storage finding", + "Types": ["Software and Configuration Checks/Synthetic"], + "Severity": {"Label": "HIGH"}, + "Workflow": {"Status": "NEW"}, + "RecordState": "ACTIVE", + "UpdatedAt": "2026-08-01T00:00:00Z", + "Resources": [ + {"Id": "arn:aws:s3:::synthetic-audit-bucket", "Type": "AwsS3Bucket"} + ], + "ProductFields": { + "cloud-audit/public_exposure": "true", + "cloud-audit/sensitive_data": "true" + } + } + ] +} + diff --git a/examples/azure-defender.json b/examples/azure-defender.json new file mode 100644 index 0000000..ddd1a14 --- /dev/null +++ b/examples/azure-defender.json @@ -0,0 +1,16 @@ +{ + "value": [ + { + "id": "/subscriptions/00000000/providers/Microsoft.Security/assessments/synthetic-001", + "name": "synthetic-001", + "type": "Microsoft.Security/assessments", + "properties": { + "status": {"code": "Unhealthy", "statusChangeDate": "2026-08-01T00:00:00Z"}, + "metadata": {"displayName": "Synthetic exposed management port", "severity": "Critical"}, + "resourceDetails": {"id": "/subscriptions/00000000/resourceGroups/synthetic/providers/Microsoft.Compute/virtualMachines/vm-001"}, + "additionalData": {"public_exposure": true, "privileged": true} + } + } + ] +} + diff --git a/examples/gcp-scc.json b/examples/gcp-scc.json new file mode 100644 index 0000000..f85ac42 --- /dev/null +++ b/examples/gcp-scc.json @@ -0,0 +1,17 @@ +{ + "listFindingsResults": [ + { + "finding": { + "name": "organizations/000000/sources/000000/findings/synthetic-001", + "resourceName": "//compute.googleapis.com/projects/synthetic/zones/us-central1-a/instances/vm-001", + "category": "PUBLIC_IP_ADDRESS", + "state": "ACTIVE", + "severity": "HIGH", + "eventTime": "2026-08-01T00:00:00Z", + "sourceProperties": {"public_exposure": true} + }, + "resource": {"type": "google.compute.Instance"} + } + ] +} + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..311ba3f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,50 @@ +[build-system] +requires = ["setuptools>=77"] +build-backend = "setuptools.build_meta" + +[project] +name = "cloud-vulnerability-audit" +version = "0.1.0" +description = "Deterministic multicloud finding normalization and risk scoring" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +authors = [{name = "Alan Nogueira"}] +keywords = ["cloud-security", "audit", "grc", "aws", "azure", "gcp"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.12", + "Topic :: Security", +] + +[project.optional-dependencies] +dev = [ + "build>=1.2", + "coverage>=7.6", + "ruff>=0.8", +] + +[project.scripts] +cloud-vulnerability-audit = "cloud_vulnerability_audit.cli:main" + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.ruff] +line-length = 100 +target-version = "py310" + +[tool.ruff.lint] +select = ["B", "E", "F", "I", "UP"] + +[tool.coverage.run] +source = ["cloud_vulnerability_audit"] + +[tool.coverage.report] +fail_under = 85 +show_missing = true diff --git a/src/cloud_vulnerability_audit/__init__.py b/src/cloud_vulnerability_audit/__init__.py new file mode 100644 index 0000000..ef9f0a1 --- /dev/null +++ b/src/cloud_vulnerability_audit/__init__.py @@ -0,0 +1,8 @@ +"""Deterministic multicloud vulnerability prioritization.""" + +from cloud_vulnerability_audit.models import Finding, ScoredFinding +from cloud_vulnerability_audit.scoring import SCORING_VERSION, score_finding + +__all__ = ["Finding", "SCORING_VERSION", "ScoredFinding", "score_finding"] +__version__ = "0.1.0" + diff --git a/src/cloud_vulnerability_audit/__main__.py b/src/cloud_vulnerability_audit/__main__.py new file mode 100644 index 0000000..2d1ad73 --- /dev/null +++ b/src/cloud_vulnerability_audit/__main__.py @@ -0,0 +1,4 @@ +from cloud_vulnerability_audit.cli import main + +raise SystemExit(main()) + diff --git a/src/cloud_vulnerability_audit/adapters.py b/src/cloud_vulnerability_audit/adapters.py new file mode 100644 index 0000000..132e6ac --- /dev/null +++ b/src/cloud_vulnerability_audit/adapters.py @@ -0,0 +1,277 @@ +"""Normalize exported findings from AWS, Azure, GCP, or the native schema.""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from typing import Any + +from cloud_vulnerability_audit.models import Finding + +PROVIDERS = {"aws", "azure", "gcp", "normalized"} + + +class UnsupportedPayloadError(ValueError): + """Raised when a provider cannot be detected safely.""" + + +def _items(payload: Any, *keys: str) -> list[dict[str, Any]]: + current = payload + for key in keys: + if not isinstance(current, Mapping): + return [] + current = current.get(key) + if isinstance(current, list): + return [item for item in current if isinstance(item, dict)] + return [] + + +def _as_bool(value: Any) -> bool | None: + if isinstance(value, bool): + return value + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"true", "yes", "1", "enabled"}: + return True + if lowered in {"false", "no", "0", "disabled"}: + return False + return None + + +def _context(mapping: Mapping[str, Any], key: str) -> Any: + aliases = { + key, + key.replace("_", "-"), + f"cloud-audit/{key}", + f"cloud_audit_{key}", + } + for alias in aliases: + if alias in mapping: + return mapping[alias] + return None + + +def _normalize_severity(value: Any) -> str: + if isinstance(value, str): + normalized = value.strip().lower().replace("-", "_") + aliases = {"informational": "informational", "info": "informational"} + if normalized in aliases: + return aliases[normalized] + if normalized in {"low", "medium", "high", "critical"}: + return normalized + if isinstance(value, (int, float)): + score = float(value) + if score <= 0: + return "informational" + if score < 40: + return "low" + if score < 70: + return "medium" + if score < 90: + return "high" + return "critical" + return "unknown" + + +def _normalize_exploitability(value: Any) -> str: + if not isinstance(value, str): + return "unknown" + normalized = value.strip().lower().replace("-", "_").replace(" ", "_") + if any(word in normalized for word in ("known", "wide", "confirmed")): + return "known_exploited" + if any(word in normalized for word in ("functional", "available", "yes", "true")): + return "functional" + if any(word in normalized for word in ("proof", "poc")): + return "proof_of_concept" + if normalized in {"none", "no", "false", "unproven"}: + return "none" + return "unknown" + + +def detect_provider(payload: Any) -> str: + if isinstance(payload, Mapping): + if isinstance(payload.get("Findings"), list) or _items(payload, "detail", "findings"): + return "aws" + if isinstance(payload.get("listFindingsResults"), list): + return "gcp" + values = payload.get("value") + if isinstance(values, list) and values: + first = values[0] + if isinstance(first, Mapping) and "properties" in first: + return "azure" + if isinstance(payload, list) and payload: + first = payload[0] + if isinstance(first, Mapping): + if "provider" in first and "severity" in first: + return "normalized" + if "SchemaVersion" in first or "ProductArn" in first: + return "aws" + if "finding" in first and isinstance(first.get("finding"), Mapping): + return "gcp" + raise UnsupportedPayloadError("provider could not be detected; pass --provider explicitly") + + +def load_findings(payload: Any, provider: str = "auto") -> list[Finding]: + selected = detect_provider(payload) if provider == "auto" else provider.lower() + if selected not in PROVIDERS: + raise UnsupportedPayloadError(f"unsupported provider: {selected}") + adapters = { + "aws": _aws, + "azure": _azure, + "gcp": _gcp, + "normalized": _normalized, + } + return adapters[selected](payload) + + +def _aws(payload: Any) -> list[Finding]: + if isinstance(payload, list): + records = [item for item in payload if isinstance(item, dict)] + else: + records = _items(payload, "Findings") or _items(payload, "detail", "findings") + findings = [] + for item in records: + severity_data = item.get("Severity") or {} + severity = _normalize_severity( + severity_data.get("Label", severity_data.get("Normalized")) + if isinstance(severity_data, Mapping) + else severity_data + ) + workflow = item.get("Workflow") or {} + workflow_status = workflow.get("Status", "") if isinstance(workflow, Mapping) else "" + record_state = str(item.get("RecordState", "")).upper() + state = ( + "resolved" + if workflow_status == "RESOLVED" or record_state == "ARCHIVED" + else "active" + ) + resources = item.get("Resources") or [] + resource = resources[0] if resources and isinstance(resources[0], Mapping) else {} + product_fields = item.get("ProductFields") or {} + user_fields = item.get("UserDefinedFields") or {} + context = {**product_fields, **user_fields} + vulnerabilities = item.get("Vulnerabilities") or [] + exploit = _context(context, "exploitability") + if exploit is None: + exploit_values = [ + vulnerability.get("ExploitAvailable") + for vulnerability in vulnerabilities + if isinstance(vulnerability, Mapping) + ] + exploit = next((value for value in exploit_values if value is not None), None) + findings.append( + Finding( + provider="aws", + finding_id=str(item.get("Id", "")), + resource_id=str(resource.get("Id", "")), + resource_type=str(resource.get("Type", "unknown")), + category=str(item.get("Types", [item.get("Title", "unknown")])[0]), + severity=severity, + state=state, + updated_at=item.get("UpdatedAt") or item.get("CreatedAt"), + exploitability=_normalize_exploitability(exploit), + public_exposure=_as_bool(_context(context, "public_exposure")), + privileged=_as_bool(_context(context, "privileged")), + sensitive_data=_as_bool(_context(context, "sensitive_data")), + compensating_control=_as_bool(_context(context, "compensating_control")), + ) + ) + return findings + + +def _azure(payload: Any) -> list[Finding]: + records = _items(payload, "value") if isinstance(payload, Mapping) else list(payload) + findings = [] + for item in records: + properties = item.get("properties") or {} + status = properties.get("status") or {} + code = str(status.get("code", "unknown")).lower() + state = "resolved" if code in {"healthy", "notapplicable", "resolved"} else "active" + metadata = properties.get("metadata") or {} + additional = properties.get("additionalData") or {} + resource = properties.get("resourceDetails") or {} + findings.append( + Finding( + provider="azure", + finding_id=str(item.get("name") or item.get("id", "")), + resource_id=str(resource.get("id") or resource.get("source", "")), + resource_type=str(resource.get("resourceType") or item.get("type", "unknown")), + category=str( + metadata.get("displayName") or properties.get("displayName", "unknown") + ), + severity=_normalize_severity(metadata.get("severity") or status.get("severity")), + state=state, + updated_at=status.get("statusChangeDate") or properties.get("timeGenerated"), + exploitability=_normalize_exploitability(_context(additional, "exploitability")), + public_exposure=_as_bool(_context(additional, "public_exposure")), + privileged=_as_bool(_context(additional, "privileged")), + sensitive_data=_as_bool(_context(additional, "sensitive_data")), + compensating_control=_as_bool(_context(additional, "compensating_control")), + ) + ) + return findings + + +def _gcp(payload: Any) -> list[Finding]: + records = ( + _items(payload, "listFindingsResults") if isinstance(payload, Mapping) else list(payload) + ) + findings = [] + for item in records: + finding = item.get("finding", item) + resource = item.get("resource") or {} + source_properties = finding.get("sourceProperties") or {} + vulnerability = finding.get("vulnerability") or {} + cve = (vulnerability.get("cve") or {}) if isinstance(vulnerability, Mapping) else {} + exploit = _context(source_properties, "exploitability") + if exploit is None and isinstance(cve, Mapping): + exploit = cve.get("exploitationActivity") + state = "resolved" if str(finding.get("state", "")).upper() == "INACTIVE" else "active" + findings.append( + Finding( + provider="gcp", + finding_id=str(finding.get("name", "")), + resource_id=str(finding.get("resourceName") or resource.get("name", "")), + resource_type=str( + resource.get("type") or source_properties.get("ResourceType", "unknown") + ), + category=str(finding.get("category", "unknown")), + severity=_normalize_severity(finding.get("severity")), + state=state, + updated_at=finding.get("eventTime") or finding.get("createTime"), + exploitability=_normalize_exploitability(exploit), + public_exposure=_as_bool(_context(source_properties, "public_exposure")), + privileged=_as_bool(_context(source_properties, "privileged")), + sensitive_data=_as_bool(_context(source_properties, "sensitive_data")), + compensating_control=_as_bool(_context(source_properties, "compensating_control")), + ) + ) + return findings + + +def _normalized(payload: Any) -> list[Finding]: + records: Iterable[Any] = ( + payload.get("findings", []) if isinstance(payload, Mapping) else payload + ) + findings = [] + for item in records: + if not isinstance(item, Mapping): + continue + findings.append( + Finding( + provider=str(item.get("provider", "normalized")).lower(), + finding_id=str(item.get("finding_id", "")), + resource_id=str(item.get("resource_id", "")), + resource_type=str(item.get("resource_type", "unknown")), + category=str(item.get("category", "unknown")), + severity=_normalize_severity(item.get("severity")), + state=str(item.get("state", "active")).lower(), + updated_at=item.get("updated_at"), + exploitability=_normalize_exploitability(item.get("exploitability")), + public_exposure=_as_bool(item.get("public_exposure")), + privileged=_as_bool(item.get("privileged")), + sensitive_data=_as_bool(item.get("sensitive_data")), + compensating_control=_as_bool(item.get("compensating_control")), + source_recognized=True, + ) + ) + return findings diff --git a/src/cloud_vulnerability_audit/cli.py b/src/cloud_vulnerability_audit/cli.py new file mode 100644 index 0000000..d2d5400 --- /dev/null +++ b/src/cloud_vulnerability_audit/cli.py @@ -0,0 +1,73 @@ +"""Command line interface.""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path + +from cloud_vulnerability_audit import __version__ +from cloud_vulnerability_audit.adapters import UnsupportedPayloadError, load_findings +from cloud_vulnerability_audit.reporting import build_report, write_json_report +from cloud_vulnerability_audit.scoring import score_finding + +MAX_INPUT_BYTES = 20 * 1024 * 1024 + + +def parser() -> argparse.ArgumentParser: + command = argparse.ArgumentParser( + prog="cloud-vulnerability-audit", + description="Normalize exported multicloud findings and calculate explainable risk scores.", + ) + command.add_argument( + "input", type=Path, help="JSON export from AWS, Azure, GCP, or native schema" + ) + command.add_argument( + "--provider", choices=("auto", "aws", "azure", "gcp", "normalized"), default="auto" + ) + command.add_argument("--output", type=Path, help="Write a privacy-aware JSON report") + command.add_argument("--version", action="version", version=__version__) + return command + + +def _load(path: Path) -> object: + if path.is_symlink(): + raise ValueError("symbolic-link inputs are not accepted") + resolved = path.expanduser().resolve(strict=True) + if not resolved.is_file(): + raise ValueError("input must be a regular file") + if resolved.stat().st_size > MAX_INPUT_BYTES: + raise ValueError("input exceeds the 20 MiB safety limit") + with resolved.open("r", encoding="utf-8") as handle: + return json.load(handle) + + +def main(argv: list[str] | None = None) -> int: + args = parser().parse_args(argv) + try: + payload = _load(args.input) + findings = load_findings(payload, args.provider) + scored = [score_finding(finding) for finding in findings] + report = build_report(scored, os.getenv("CLOUD_AUDIT_HMAC_SECRET")) + if args.output: + write_json_report(args.output, report) + summary = { + key: report[key] + for key in ( + "environment_score", + "environment_level", + "finding_count", + "counts_by_level", + ) + } + print(json.dumps(summary, ensure_ascii=False)) + return 0 + except (OSError, ValueError, json.JSONDecodeError, UnsupportedPayloadError) as error: + print(f"error: {error}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/cloud_vulnerability_audit/models.py b/src/cloud_vulnerability_audit/models.py new file mode 100644 index 0000000..4dc30d7 --- /dev/null +++ b/src/cloud_vulnerability_audit/models.py @@ -0,0 +1,33 @@ +"""Internal normalized models. Raw provider payloads are never reported.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class Finding: + provider: str + finding_id: str + resource_id: str + resource_type: str + category: str + severity: str + state: str + updated_at: str | None = None + exploitability: str = "unknown" + public_exposure: bool | None = None + privileged: bool | None = None + sensitive_data: bool | None = None + compensating_control: bool | None = None + source_recognized: bool = True + + +@dataclass(frozen=True) +class ScoredFinding: + finding: Finding + risk_score: int + risk_level: str + confidence_score: int + factors: tuple[str, ...] = field(default_factory=tuple) + diff --git a/src/cloud_vulnerability_audit/reporting.py b/src/cloud_vulnerability_audit/reporting.py new file mode 100644 index 0000000..67e5cfb --- /dev/null +++ b/src/cloud_vulnerability_audit/reporting.py @@ -0,0 +1,77 @@ +"""Privacy-aware report construction and atomic JSON output.""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from tempfile import NamedTemporaryFile + +from cloud_vulnerability_audit.models import ScoredFinding +from cloud_vulnerability_audit.scoring import SCORING_VERSION, environment_score + + +def _reference(value: str, secret: str | None = None) -> str: + encoded = value.encode("utf-8") + digest = ( + hmac.new(secret.encode("utf-8"), encoded, hashlib.sha256).hexdigest() + if secret + else hashlib.sha256(encoded).hexdigest() + ) + return digest[:16] + + +def build_report(findings: list[ScoredFinding], secret: str | None = None) -> dict: + score, level = environment_score(findings) + counts = {name: 0 for name in ("informational", "low", "medium", "high", "critical")} + rows = [] + for scored in findings: + counts[scored.risk_level] += 1 + raw = scored.finding + rows.append( + { + "provider": raw.provider, + "finding_ref": _reference(raw.finding_id, secret), + "resource_ref": _reference(raw.resource_id, secret), + "resource_type": raw.resource_type, + "category": raw.category, + "provider_severity": raw.severity, + "state": raw.state, + "risk_score": scored.risk_score, + "risk_level": scored.risk_level, + "confidence_score": scored.confidence_score, + "factors": list(scored.factors), + } + ) + return { + "schema_version": "cloud-audit-report-1.0", + "scoring_version": SCORING_VERSION, + "generated_at": datetime.now(timezone.utc).isoformat(), + "environment_score": score, + "environment_level": level, + "finding_count": len(findings), + "counts_by_level": counts, + "privacy_notice": "identifiers are pseudonymized; raw titles and descriptions are omitted", + "findings": rows, + } + + +def write_json_report(path: Path, report: dict) -> None: + destination = path.expanduser().resolve() + destination.parent.mkdir(parents=True, exist_ok=True) + with NamedTemporaryFile( + "w", encoding="utf-8", dir=destination.parent, delete=False, prefix=".cloud-audit-" + ) as temporary: + json.dump(report, temporary, ensure_ascii=False, indent=2) + temporary.write("\n") + temporary_path = Path(temporary.name) + try: + os.chmod(temporary_path, 0o600) + temporary_path.replace(destination) + except Exception: + temporary_path.unlink(missing_ok=True) + raise + diff --git a/src/cloud_vulnerability_audit/scoring.py b/src/cloud_vulnerability_audit/scoring.py new file mode 100644 index 0000000..c0c05ee --- /dev/null +++ b/src/cloud_vulnerability_audit/scoring.py @@ -0,0 +1,135 @@ +"""Explainable scoring for normalized cloud findings.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from statistics import mean + +from cloud_vulnerability_audit.models import Finding, ScoredFinding + +SCORING_VERSION = "cloud-risk-1.0.0" +SEVERITY_POINTS = { + "informational": 0, + "low": 10, + "medium": 25, + "high": 45, + "critical": 70, + "unknown": 15, +} +EXPLOIT_POINTS = { + "unknown": 0, + "none": 0, + "proof_of_concept": 5, + "functional": 10, + "known_exploited": 20, +} +INACTIVE_STATES = {"archived", "closed", "healthy", "inactive", "not_applicable", "resolved"} + + +def risk_level(score: int) -> str: + if score == 0: + return "informational" + if score < 20: + return "low" + if score < 40: + return "medium" + if score < 70: + return "high" + return "critical" + + +def _parse_timestamp(value: str | None) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _confidence(finding: Finding) -> int: + score = 30 if finding.source_recognized else 0 + score += 25 if finding.severity != "unknown" else 0 + score += 15 if finding.state not in {"", "unknown"} else 0 + score += 10 if finding.resource_id else 0 + score += 10 if finding.finding_id else 0 + score += 5 if _parse_timestamp(finding.updated_at) else 0 + context = ( + finding.exploitability != "unknown" + or finding.public_exposure is not None + or finding.privileged is not None + or finding.sensitive_data is not None + or finding.compensating_control is not None + ) + score += 5 if context else 0 + return min(score, 100) + + +def score_finding(finding: Finding, now: datetime | None = None) -> ScoredFinding: + confidence = _confidence(finding) + if finding.state.lower() in INACTIVE_STATES: + return ScoredFinding( + finding=finding, + risk_score=0, + risk_level="informational", + confidence_score=confidence, + factors=("inactive_or_resolved=0",), + ) + + factors = [] + points = SEVERITY_POINTS.get(finding.severity, SEVERITY_POINTS["unknown"]) + factors.append(f"severity:{finding.severity}={points}") + + exploit_points = EXPLOIT_POINTS.get(finding.exploitability, 0) + if exploit_points: + points += exploit_points + factors.append(f"exploitability:{finding.exploitability}=+{exploit_points}") + if finding.public_exposure is True: + points += 15 + factors.append("public_exposure=+15") + if finding.privileged is True: + points += 10 + factors.append("privileged=+10") + if finding.sensitive_data is True: + points += 10 + factors.append("sensitive_data=+10") + + reference = now or datetime.now(timezone.utc) + updated = _parse_timestamp(finding.updated_at) + if updated: + age_days = max((reference.astimezone(timezone.utc) - updated).days, 0) + if age_days > 180: + points += 10 + factors.append("age_over_180_days=+10") + elif age_days > 90: + points += 5 + factors.append("age_over_90_days=+5") + + if finding.compensating_control is True: + points -= 10 + factors.append("compensating_control=-10") + + points = max(0, min(points, 100)) + return ScoredFinding( + finding=finding, + risk_score=points, + risk_level=risk_level(points), + confidence_score=confidence, + factors=tuple(factors), + ) + + +def environment_score(findings: list[ScoredFinding]) -> tuple[int, str]: + active_scores = sorted( + (finding.risk_score for finding in findings if finding.risk_score > 0), reverse=True + ) + if not active_scores: + return 0, "informational" + top_scores = active_scores[:10] + critical_ratio = sum(score >= 70 for score in active_scores) / len(active_scores) * 100 + score = round(0.50 * active_scores[0] + 0.35 * mean(top_scores) + 0.15 * critical_ratio) + score = min(score, 100) + return score, risk_level(score) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..905593d --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Synthetic test suite.""" diff --git a/tests/test_adapters.py b/tests/test_adapters.py new file mode 100644 index 0000000..0828e18 --- /dev/null +++ b/tests/test_adapters.py @@ -0,0 +1,84 @@ +from unittest import TestCase + +from cloud_vulnerability_audit.adapters import ( + UnsupportedPayloadError, + detect_provider, + load_findings, +) + + +class AdapterTests(TestCase): + def test_aws_security_hub_asff(self): + payload = { + "Findings": [ + { + "SchemaVersion": "2018-10-08", + "Id": "aws-finding", + "Title": "Synthetic public bucket", + "Types": ["Software and Configuration Checks/Synthetic"], + "Severity": {"Label": "HIGH"}, + "Workflow": {"Status": "NEW"}, + "RecordState": "ACTIVE", + "UpdatedAt": "2026-08-01T00:00:00Z", + "Resources": [{"Id": "arn:synthetic", "Type": "AwsS3Bucket"}], + "ProductFields": {"cloud-audit/public_exposure": "true"}, + } + ] + } + self.assertEqual(detect_provider(payload), "aws") + result = load_findings(payload) + self.assertEqual(result[0].severity, "high") + self.assertTrue(result[0].public_exposure) + + def test_azure_defender_assessment(self): + payload = { + "value": [ + { + "id": "/subscriptions/synthetic/providers/Microsoft.Security/assessments/a1", + "name": "a1", + "type": "Microsoft.Security/assessments", + "properties": { + "status": {"code": "Unhealthy"}, + "metadata": {"displayName": "Synthetic", "severity": "Critical"}, + "resourceDetails": {"id": "/subscriptions/synthetic/vm1"}, + "additionalData": {"public_exposure": True}, + }, + } + ] + } + self.assertEqual(detect_provider(payload), "azure") + result = load_findings(payload) + self.assertEqual(result[0].severity, "critical") + self.assertEqual(result[0].state, "active") + + def test_gcp_security_command_center(self): + payload = { + "listFindingsResults": [ + { + "finding": { + "name": "organizations/0/sources/0/findings/synthetic", + "resourceName": "//compute.googleapis.com/projects/synthetic/instances/vm1", + "category": "PUBLIC_IP_ADDRESS", + "state": "ACTIVE", + "severity": "HIGH", + "eventTime": "2026-08-01T00:00:00Z", + "sourceProperties": {"public_exposure": True}, + }, + "resource": {"type": "google.compute.Instance"}, + } + ] + } + self.assertEqual(detect_provider(payload), "gcp") + result = load_findings(payload) + self.assertEqual(result[0].resource_type, "google.compute.Instance") + self.assertTrue(result[0].public_exposure) + + def test_normalized_schema(self): + payload = [{"provider": "aws", "severity": "medium", "state": "active"}] + self.assertEqual(detect_provider(payload), "normalized") + self.assertEqual(load_findings(payload)[0].severity, "medium") + + def test_unknown_payload_is_rejected(self): + with self.assertRaises(UnsupportedPayloadError): + detect_provider({"unexpected": []}) + diff --git a/tests/test_reporting_cli.py b/tests/test_reporting_cli.py new file mode 100644 index 0000000..e33a92e --- /dev/null +++ b/tests/test_reporting_cli.py @@ -0,0 +1,77 @@ +import json +import stat +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import TestCase, mock + +from cloud_vulnerability_audit.cli import main +from cloud_vulnerability_audit.models import Finding +from cloud_vulnerability_audit.reporting import build_report, write_json_report +from cloud_vulnerability_audit.scoring import score_finding + + +class ReportingTests(TestCase): + def test_report_omits_raw_identifiers_and_titles(self): + raw_identifier = "arn:aws:s3:::confidential-resource-name" + scored = score_finding( + Finding( + provider="aws", + finding_id="confidential-finding-id", + resource_id=raw_identifier, + resource_type="AwsS3Bucket", + category="Synthetic category", + severity="high", + state="active", + ) + ) + rendered = json.dumps(build_report([scored])) + self.assertNotIn(raw_identifier, rendered) + self.assertNotIn("confidential-finding-id", rendered) + self.assertIn("resource_ref", rendered) + + def test_hmac_secret_changes_reference(self): + scored = score_finding( + Finding("aws", "id", "resource", "type", "category", "low", "active") + ) + first = build_report([scored], "secret-one") + second = build_report([scored], "secret-two") + self.assertNotEqual( + first["findings"][0]["resource_ref"], second["findings"][0]["resource_ref"] + ) + + def test_atomic_report_is_private_on_posix(self): + with TemporaryDirectory() as directory: + path = Path(directory) / "report.json" + write_json_report(path, {"ok": True}) + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + + def test_cli_processes_synthetic_export(self): + payload = [ + { + "provider": "aws", + "finding_id": "f1", + "resource_id": "r1", + "resource_type": "AwsS3Bucket", + "category": "synthetic", + "severity": "high", + "state": "active", + "public_exposure": True, + } + ] + with TemporaryDirectory() as directory: + source = Path(directory) / "input.json" + destination = Path(directory) / "report.json" + source.write_text(json.dumps(payload), encoding="utf-8") + with mock.patch("builtins.print") as output: + code = main([str(source), "--provider", "normalized", "--output", str(destination)]) + self.assertEqual(code, 0) + self.assertTrue(destination.exists()) + self.assertIn("environment_score", output.call_args.args[0]) + + def test_cli_rejects_invalid_json(self): + with TemporaryDirectory() as directory: + source = Path(directory) / "input.json" + source.write_text("not-json", encoding="utf-8") + with mock.patch("sys.stderr"): + self.assertEqual(main([str(source)]), 2) + diff --git a/tests/test_scoring.py b/tests/test_scoring.py new file mode 100644 index 0000000..79e4ac5 --- /dev/null +++ b/tests/test_scoring.py @@ -0,0 +1,70 @@ +from datetime import datetime, timezone +from unittest import TestCase + +from cloud_vulnerability_audit.models import Finding +from cloud_vulnerability_audit.scoring import environment_score, score_finding + + +def finding(**changes): + values = { + "provider": "aws", + "finding_id": "finding-1", + "resource_id": "arn:aws:s3:::synthetic-bucket", + "resource_type": "AwsS3Bucket", + "category": "synthetic", + "severity": "high", + "state": "active", + "updated_at": "2025-01-01T00:00:00Z", + } + values.update(changes) + return Finding(**values) + + +class ScoringTests(TestCase): + def setUp(self): + self.now = datetime(2026, 8, 21, tzinfo=timezone.utc) + + def test_critical_context_is_capped_at_100(self): + scored = score_finding( + finding( + severity="critical", + exploitability="known_exploited", + public_exposure=True, + privileged=True, + sensitive_data=True, + ), + self.now, + ) + self.assertEqual(scored.risk_score, 100) + self.assertEqual(scored.risk_level, "critical") + self.assertIn("public_exposure=+15", scored.factors) + + def test_resolved_finding_has_zero_risk(self): + scored = score_finding(finding(state="resolved", severity="critical"), self.now) + self.assertEqual(scored.risk_score, 0) + self.assertEqual(scored.factors, ("inactive_or_resolved=0",)) + + def test_compensating_control_reduces_risk(self): + without = score_finding(finding(updated_at=None), self.now) + with_control = score_finding(finding(updated_at=None, compensating_control=True), self.now) + self.assertEqual(without.risk_score - with_control.risk_score, 10) + + def test_confidence_is_separate_from_risk(self): + complete = score_finding(finding(public_exposure=False), self.now) + incomplete = score_finding( + finding(finding_id="", resource_id="", severity="unknown", updated_at=None), self.now + ) + self.assertGreater(complete.confidence_score, incomplete.confidence_score) + self.assertNotEqual(complete.risk_score, complete.confidence_score) + + def test_environment_score_emphasizes_top_risk(self): + scored = [ + score_finding(finding(finding_id="1", severity="critical", updated_at=None), self.now), + score_finding(finding(finding_id="2", severity="low", updated_at=None), self.now), + ] + score, level = environment_score(scored) + self.assertEqual(score, 56) + self.assertEqual(level, "high") + + def test_empty_environment_is_informational(self): + self.assertEqual(environment_score([]), (0, "informational"))