generated from coulomb/repo-seed
386 lines
15 KiB
Python
386 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .discovery import short_fingerprint
|
|
from .loader import load_yaml, repo_root
|
|
from .schema_validation import draft202012_validator
|
|
|
|
|
|
EXTRACTOR_VERSION = "0.1.0"
|
|
DEFAULT_ROOT_MANIFEST_PATH = repo_root() / "fabric" / "discovery" / "railiance-accountability-roots.yaml"
|
|
|
|
|
|
def load_accountability_root_manifest(path: Path | None = None, *, validate: bool = True) -> dict[str, Any]:
|
|
manifest_path = path or DEFAULT_ROOT_MANIFEST_PATH
|
|
manifest = load_yaml(manifest_path)
|
|
if not isinstance(manifest, dict):
|
|
raise ValueError(f"accountability root manifest must be a mapping: {manifest_path}")
|
|
if validate:
|
|
validator = draft202012_validator(repo_root() / "schemas" / "accountability-root-manifest.schema.yaml")
|
|
errors = sorted(validator.iter_errors(manifest), key=lambda error: list(error.path))
|
|
if errors:
|
|
location = ".".join(str(part) for part in errors[0].path) or "<root>"
|
|
raise ValueError(f"invalid accountability root manifest at {location}: {errors[0].message}")
|
|
return manifest
|
|
|
|
|
|
def collect_accountability_root_evidence(
|
|
manifest_path: Path | None = None,
|
|
*,
|
|
include_remote: bool = False,
|
|
max_items_per_root: int = 200,
|
|
) -> dict[str, Any]:
|
|
manifest_path = manifest_path or DEFAULT_ROOT_MANIFEST_PATH
|
|
manifest = load_accountability_root_manifest(manifest_path)
|
|
generated_at = _utc_now()
|
|
roots: list[dict[str, Any]] = []
|
|
review_artifacts: list[dict[str, Any]] = []
|
|
|
|
for root in manifest.get("discovery_roots", []):
|
|
if not isinstance(root, dict):
|
|
continue
|
|
root_record = {
|
|
"root_id": root.get("id", ""),
|
|
"root_type": root.get("type", ""),
|
|
"status": root.get("status", "planned"),
|
|
"fabric_id": root.get("fabric_id", ""),
|
|
"owner_actor_id": root.get("owner_actor_id", ""),
|
|
"safe_discovery": _source(root).get("safe_discovery", "metadata_only"),
|
|
"evidence": [],
|
|
}
|
|
if root.get("subfabric_id"):
|
|
root_record["subfabric_id"] = root["subfabric_id"]
|
|
try:
|
|
root_record["evidence"] = _collect_root_evidence(
|
|
root,
|
|
include_remote=include_remote,
|
|
max_items=max_items_per_root,
|
|
)
|
|
except Exception as exc: # pragma: no cover - defensive boundary for operator runs
|
|
review_artifacts.append(
|
|
_review_artifact(
|
|
root,
|
|
"adapter_failed",
|
|
"error",
|
|
f"{type(exc).__name__}: {exc}",
|
|
)
|
|
)
|
|
roots.append(root_record)
|
|
|
|
result = {
|
|
"apiVersion": "railiance.fabric/v1alpha2",
|
|
"kind": "AccountabilityRootEvidenceRun",
|
|
"generated_at": generated_at,
|
|
"manifest": {
|
|
"id": manifest.get("metadata", {}).get("id", ""),
|
|
"path": _display_path(manifest_path),
|
|
"fingerprint": _file_sha256(manifest_path) or short_fingerprint(manifest),
|
|
},
|
|
"roots": roots,
|
|
"review_artifacts": review_artifacts,
|
|
}
|
|
validator = draft202012_validator(repo_root() / "schemas" / "accountability-root-evidence.schema.yaml")
|
|
errors = sorted(validator.iter_errors(result), key=lambda error: list(error.path))
|
|
if errors:
|
|
location = ".".join(str(part) for part in errors[0].path) or "<root>"
|
|
raise ValueError(f"invalid accountability root evidence at {location}: {errors[0].message}")
|
|
return result
|
|
|
|
|
|
def _collect_root_evidence(root: dict[str, Any], *, include_remote: bool, max_items: int) -> list[dict[str, Any]]:
|
|
root_type = str(root.get("type") or "")
|
|
if root.get("status") == "disabled":
|
|
return [_declared_evidence(root, "root_disabled", "skipped", "Discovery root is disabled.")]
|
|
if root_type == "registry_manifest":
|
|
return _registry_manifest_evidence(root, max_items=max_items)
|
|
if root_type == "repository_checkout":
|
|
return _repository_checkout_evidence(root)
|
|
if root_type == "host_path":
|
|
return _glob_root_evidence(root, "host_path_match", max_items=max_items)
|
|
if root_type in {"deployment_automation", "infrastructure_manifest", "service_config", "endpoint_contract"}:
|
|
return _glob_root_evidence(root, root_type, max_items=max_items)
|
|
if root_type == "state_hub_repo_inventory":
|
|
return _state_hub_evidence(root, include_remote=include_remote)
|
|
if root_type in {"gitea_organization", "gitea_repository"}:
|
|
return [_declared_evidence(root, root_type, "declared", f"{root_type} root declared.")]
|
|
if root_type in {"secret_root", "backup_recovery", "manual_review_queue"}:
|
|
return _metadata_root_evidence(root)
|
|
return [_declared_evidence(root, root_type or "unknown_root", "declared", "Discovery root declared.")]
|
|
|
|
|
|
def _registry_manifest_evidence(root: dict[str, Any], *, max_items: int) -> list[dict[str, Any]]:
|
|
source = _source(root)
|
|
manifest_path = _resolve_path(source.get("manifest_path") or source.get("path"))
|
|
if not manifest_path.exists():
|
|
return [_declared_evidence(root, "registry_manifest_missing", "unavailable", f"Manifest missing: {manifest_path}")]
|
|
manifest = load_yaml(manifest_path)
|
|
repositories = manifest.get("repositories") if isinstance(manifest, dict) else []
|
|
if not isinstance(repositories, list):
|
|
return [_declared_evidence(root, "registry_manifest_invalid", "unavailable", "Manifest has no repositories list.")]
|
|
|
|
evidence: list[dict[str, Any]] = [
|
|
_file_evidence(root, manifest_path, "registry_manifest", summary=f"Registry manifest with {len(repositories)} repositories.")
|
|
]
|
|
for index, repo in enumerate(repositories[:max_items]):
|
|
if not isinstance(repo, dict):
|
|
continue
|
|
repo_source = {
|
|
"manifest_path": _display_path(manifest_path),
|
|
"json_pointer": f"/repositories/{index}",
|
|
"repo_slug": repo.get("slug", ""),
|
|
"path": repo.get("path", ""),
|
|
"remote_url": repo.get("remote_url", ""),
|
|
}
|
|
attributes = {
|
|
"name": repo.get("name", ""),
|
|
"domain": repo.get("domain", ""),
|
|
"default_branch": repo.get("default_branch", ""),
|
|
"state_hub_repo_id": repo.get("state_hub_repo_id", ""),
|
|
"has_local_path": bool(repo.get("path")),
|
|
"has_remote_url": bool(repo.get("remote_url")),
|
|
}
|
|
evidence.append(
|
|
_evidence_item(
|
|
root,
|
|
evidence_type="registered_repository",
|
|
state="declared",
|
|
source=repo_source,
|
|
summary=f"Registered repository {repo.get('slug', '<unknown>')}.",
|
|
attributes={key: value for key, value in attributes.items() if value not in ("", None)},
|
|
)
|
|
)
|
|
if len(repositories) > max_items:
|
|
evidence.append(_declared_evidence(root, "registry_manifest_truncated", "skipped", f"Skipped {len(repositories) - max_items} repositories beyond max_items_per_root."))
|
|
return evidence
|
|
|
|
|
|
def _repository_checkout_evidence(root: dict[str, Any]) -> list[dict[str, Any]]:
|
|
source = _source(root)
|
|
checkout = _resolve_path(source.get("path"))
|
|
if not checkout.exists():
|
|
return [_declared_evidence(root, "repository_checkout_missing", "unavailable", f"Checkout missing: {checkout}")]
|
|
attributes = {
|
|
"repo_slug": source.get("repo_slug", ""),
|
|
"path_exists": True,
|
|
"has_git_dir": (checkout / ".git").exists(),
|
|
"has_fabric_dir": (checkout / "fabric").exists(),
|
|
"remote_origin": _git_value(checkout, "config", "--get", "remote.origin.url") or source.get("remote_url", ""),
|
|
"head": _git_value(checkout, "rev-parse", "HEAD") or "",
|
|
"branch": _git_value(checkout, "rev-parse", "--abbrev-ref", "HEAD") or "",
|
|
}
|
|
return [
|
|
_evidence_item(
|
|
root,
|
|
evidence_type="repository_checkout",
|
|
state="observed",
|
|
source={"path": _display_path(checkout), "repo_slug": source.get("repo_slug", "")},
|
|
summary=f"Repository checkout observed at {_display_path(checkout)}.",
|
|
attributes={key: value for key, value in attributes.items() if value not in ("", None)},
|
|
)
|
|
]
|
|
|
|
|
|
def _glob_root_evidence(root: dict[str, Any], evidence_type: str, *, max_items: int) -> list[dict[str, Any]]:
|
|
source = _source(root)
|
|
base = _resolve_path(source.get("path") or ".")
|
|
patterns = source.get("patterns") if isinstance(source.get("patterns"), list) else ["*"]
|
|
if not base.exists():
|
|
return [_declared_evidence(root, f"{evidence_type}_missing", "unavailable", f"Root path missing: {base}")]
|
|
matches: list[Path] = []
|
|
for pattern in patterns:
|
|
matches.extend(sorted(base.glob(str(pattern))))
|
|
if len(matches) >= max_items:
|
|
break
|
|
evidence = [
|
|
_evidence_item(
|
|
root,
|
|
evidence_type=evidence_type,
|
|
state="observed",
|
|
source={"path": _display_path(path)},
|
|
summary=f"Observed {evidence_type} at {_display_path(path)}.",
|
|
attributes=_file_attributes(path),
|
|
)
|
|
for path in matches[:max_items]
|
|
]
|
|
if not evidence:
|
|
evidence.append(_declared_evidence(root, f"{evidence_type}_empty", "unavailable", f"No files matched under {base}."))
|
|
if len(matches) > max_items:
|
|
evidence.append(_declared_evidence(root, f"{evidence_type}_truncated", "skipped", f"Skipped {len(matches) - max_items} matches beyond max_items_per_root."))
|
|
return evidence
|
|
|
|
|
|
def _state_hub_evidence(root: dict[str, Any], *, include_remote: bool) -> list[dict[str, Any]]:
|
|
source = _source(root)
|
|
if not include_remote:
|
|
return [_declared_evidence(root, "state_hub_repo_inventory", "declared", "State Hub repo inventory root declared; remote fetch disabled.")]
|
|
base_url = str(source.get("base_url") or "").rstrip("/")
|
|
evidence: list[dict[str, Any]] = []
|
|
for api_path in source.get("api_paths") or ["/managed-repos/"]:
|
|
url = f"{base_url}{api_path}"
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=5) as response:
|
|
payload = json.loads(response.read())
|
|
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
|
|
evidence.append(_declared_evidence(root, "state_hub_fetch_failed", "unavailable", f"{url}: {exc}"))
|
|
continue
|
|
count = len(payload) if isinstance(payload, list) else len(payload.get("items", [])) if isinstance(payload, dict) else 0
|
|
evidence.append(
|
|
_evidence_item(
|
|
root,
|
|
evidence_type="state_hub_repo_inventory",
|
|
state="observed",
|
|
source={"url": url},
|
|
summary=f"Fetched State Hub repository inventory from {url}.",
|
|
attributes={"item_count": count, "payload_fingerprint": short_fingerprint(payload)},
|
|
)
|
|
)
|
|
return evidence
|
|
|
|
|
|
def _metadata_root_evidence(root: dict[str, Any]) -> list[dict[str, Any]]:
|
|
source = _source(root)
|
|
path = source.get("path")
|
|
if path:
|
|
resolved = _resolve_path(path)
|
|
if resolved.exists():
|
|
return [_file_evidence(root, resolved, str(root.get("type") or "metadata_root"))]
|
|
return [_declared_evidence(root, str(root.get("type") or "metadata_root"), "planned" if root.get("status") == "planned" else "declared", "Metadata-only root declared.")]
|
|
|
|
|
|
def _file_evidence(root: dict[str, Any], path: Path, evidence_type: str, *, summary: str | None = None) -> dict[str, Any]:
|
|
return _evidence_item(
|
|
root,
|
|
evidence_type=evidence_type,
|
|
state="observed",
|
|
source={"path": _display_path(path)},
|
|
summary=summary or f"Observed {evidence_type} file at {_display_path(path)}.",
|
|
attributes=_file_attributes(path),
|
|
)
|
|
|
|
|
|
def _declared_evidence(root: dict[str, Any], evidence_type: str, state: str, summary: str) -> dict[str, Any]:
|
|
source = _source(root)
|
|
return _evidence_item(
|
|
root,
|
|
evidence_type=evidence_type,
|
|
state=state,
|
|
source={key: value for key, value in source.items() if key != "safe_discovery"},
|
|
summary=summary,
|
|
attributes={"safe_discovery": source.get("safe_discovery", "metadata_only")},
|
|
)
|
|
|
|
|
|
def _evidence_item(
|
|
root: dict[str, Any],
|
|
*,
|
|
evidence_type: str,
|
|
state: str,
|
|
source: dict[str, Any],
|
|
summary: str,
|
|
attributes: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
payload = {
|
|
"root_id": root.get("id", ""),
|
|
"evidence_type": evidence_type,
|
|
"state": state,
|
|
"source": source,
|
|
"summary": summary,
|
|
"attributes": attributes or {},
|
|
}
|
|
fingerprint = short_fingerprint(payload, length=16)
|
|
return {
|
|
"id": f"evidence:{root.get('id', 'root')}:{fingerprint}",
|
|
"root_id": root.get("id", ""),
|
|
"evidence_type": evidence_type,
|
|
"state": state,
|
|
"durable": True,
|
|
"live_telemetry": False,
|
|
"source": source,
|
|
"provenance": {
|
|
"extractor_id": "accountability-root-adapter",
|
|
"extractor_version": EXTRACTOR_VERSION,
|
|
"method": "deterministic",
|
|
"origin": "deterministic",
|
|
},
|
|
"fingerprint": fingerprint,
|
|
"summary": summary,
|
|
"attributes": attributes or {},
|
|
}
|
|
|
|
|
|
def _review_artifact(root: dict[str, Any], artifact_type: str, severity: str, message: str) -> dict[str, Any]:
|
|
return {
|
|
"root_id": root.get("id", ""),
|
|
"artifact_type": artifact_type,
|
|
"severity": severity,
|
|
"message": message,
|
|
"source": _source(root),
|
|
}
|
|
|
|
|
|
def _source(root: dict[str, Any]) -> dict[str, Any]:
|
|
source = root.get("source")
|
|
return source if isinstance(source, dict) else {}
|
|
|
|
|
|
def _resolve_path(value: object) -> Path:
|
|
path = Path(str(value or "."))
|
|
return path if path.is_absolute() else repo_root() / path
|
|
|
|
|
|
def _display_path(path: Path) -> str:
|
|
try:
|
|
return path.resolve().relative_to(repo_root()).as_posix()
|
|
except ValueError:
|
|
return str(path.resolve())
|
|
|
|
|
|
def _file_attributes(path: Path) -> dict[str, Any]:
|
|
attributes: dict[str, Any] = {
|
|
"path_type": "directory" if path.is_dir() else "file",
|
|
"exists": path.exists(),
|
|
}
|
|
if path.is_file():
|
|
attributes["size_bytes"] = path.stat().st_size
|
|
attributes["sha256"] = _file_sha256(path)
|
|
return attributes
|
|
|
|
|
|
def _file_sha256(path: Path) -> str | None:
|
|
if not path.is_file():
|
|
return None
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _git_value(repo_path: Path, *args: str) -> str | None:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
cwd=repo_path,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
timeout=5,
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
value = result.stdout.strip()
|
|
return value or None
|
|
|
|
|
|
def _utc_now() -> str:
|
|
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|