From 36a2517e97b6a8277e7df0732f09afbfbbd48a26 Mon Sep 17 00:00:00 2001 From: Shaun Smith <51304449+smith-xyz@users.noreply.github.com> Date: Wed, 21 Oct 2026 12:33:50 -0400 Subject: [PATCH 1/2] feat(portfolio): repository observation onto the contract tables Takes over PR #5: girish's observation flow, re-homed onto the table repositories and keyed by repository_id. No legacy bridges - targets, audits and open counts read the contract tables and the finding_status view; observations persist in memory until contracts lands repository_observation. Includes Michele's required forge-parser corrections with regression tests, the per-role observation modules (models/observation, services/observation, repositories/observation), and a repo commit-prep skill. --- .agents/skills/commit-prep/SKILL.md | 58 ++ README.md | 1 + src/traust_core/v1/context/config.py | 2 + src/traust_core/v1/context/context.py | 61 +- src/traust_core/v1/context/observation.py | 34 + src/traust_core/v1/models/assessment.py | 6 +- src/traust_core/v1/models/inventory.py | 3 + src/traust_core/v1/models/observation.py | 118 +++ src/traust_core/v1/providers/forge.py | 231 ++++++ src/traust_core/v1/providers/sources.py | 32 + src/traust_core/v1/repositories/contract.py | 1 + .../v1/repositories/designations.py | 44 ++ .../v1/repositories/observation.py | 33 + .../repositories/traust/findings/__init__.py | 4 + .../v1/repositories/traust/findings/status.py | 47 ++ src/traust_core/v1/services/assessment.py | 90 ++- src/traust_core/v1/services/observation.py | 244 ++++++ src/traust_core/v1/services/portfolio.py | 178 ++++- .../observations/recorded-github.json | 31 + tests/fixtures/observations/reference.json | 727 ++++++++++++++++++ tests/v1/context/test_observation.py | 24 + tests/v1/observation_support.py | 143 ++++ tests/v1/providers/test_forge.py | 140 ++++ .../tables/test_observation_target.py | 101 +++ tests/v1/repositories/test_designations.py | 41 + .../v1/services/test_portfolio_observation.py | 125 +++ 26 files changed, 2506 insertions(+), 13 deletions(-) create mode 100644 .agents/skills/commit-prep/SKILL.md create mode 100644 src/traust_core/v1/context/observation.py create mode 100644 src/traust_core/v1/models/observation.py create mode 100644 src/traust_core/v1/providers/forge.py create mode 100644 src/traust_core/v1/repositories/designations.py create mode 100644 src/traust_core/v1/repositories/observation.py create mode 100644 src/traust_core/v1/repositories/traust/findings/status.py create mode 100644 src/traust_core/v1/services/observation.py create mode 100644 tests/fixtures/observations/recorded-github.json create mode 100644 tests/fixtures/observations/reference.json create mode 100644 tests/v1/context/test_observation.py create mode 100644 tests/v1/observation_support.py create mode 100644 tests/v1/providers/test_forge.py create mode 100644 tests/v1/repositories/tables/test_observation_target.py create mode 100644 tests/v1/repositories/test_designations.py create mode 100644 tests/v1/services/test_portfolio_observation.py diff --git a/.agents/skills/commit-prep/SKILL.md b/.agents/skills/commit-prep/SKILL.md new file mode 100644 index 00000000..d6842b95 --- /dev/null +++ b/.agents/skills/commit-prep/SKILL.md @@ -0,0 +1,58 @@ +--- +name: commit-prep +description: Take traust-core changes from working tree to merged PR in one run — gates, conventional signed commit, branch, PR body, auto-merge. Use when asked to prep a commit, open a PR, or land changes in traust-core. +--- + +# Commit Prep (traust-core) + +Run the changes through this repo's gates, then land them as a PR. The org ruleset +requires a PR for main but **zero approvals** — self-merge is legal — and commits +must be signed (`commit.gpgsign` is on in this repo, so signing is automatic). + +## Workflow + +1. **Scope the change.** `git status`, `git diff --stat`, `git log --oneline -5` for + style. If the tree mixes unrelated changes, split commits by concern before + proceeding — one PR carries one story. +2. **Run the gates, in order, all must pass:** + ```bash + make lint-fix && make test && make coverage && make db-up && make test-integration + ``` + - `make coverage` fails below 80% — new code needs tests; stub bodies are excluded. + - **Known flake:** integration fixtures failing with `UndefinedColumn` (e.g. + `column "severity" does not exist`) means the shared `traust_test` container + holds the other DDL generation (contracts branch vs the pinned dep both + recreate it). Clear and re-run: + ```bash + podman exec traust-postgres psql -U traust -d traust_test -c \ + 'DROP SCHEMA IF EXISTS traust_storage CASCADE; DROP SCHEMA IF EXISTS traust_ledger CASCADE;' + ``` + If it persists after clearing, it is a real schema mismatch — stop and report. +3. **Commit.** Conventional subject, enforced by the commit-msg hook: + `type(scope): subject` — types feat/fix/refactor/docs/test/chore/ci/perf, scope is + the narrowest affected area, imperative lowercase subject. Hooks run ruff on + staged files; do not bypass with `--no-verify`. +4. **Branch and push.** Kebab-case `type/short-description`, off current main: + `git push -u origin `. Never force-push `main` (org ruleset blocks it). +5. **Open the PR** with `gh pr create --base main`, body shaped like: + - **What** — the change in 1–3 bullets, tied to the model/slice it serves + - **The decisions** — any rule that governed it (one owner, DB-owned identity, pattern set) + - **Tests** — which gates ran, any known environmental caveat + - **Tracker** — the XWING story key +6. **Merge.** `gh pr merge --merge --delete-branch --auto` — merges when the five + CI checks pass (lint, test 3.11/3.12, api-check, postgres). Post-merge: pull main + locally, log the milestone to the story. + +## Rules + +- Gates first: no PR opens with a red suite. If a gate cannot pass, report why — + do not weaken the gate. +- One PR, one story. Split mixed trees into separate commits/PRs. +- Never merge another author's PR or one with a human's open review request. +- Never push directly to `main` — the ruleset rejects it anyway. +- If contracts or DDL shapes are involved, say so in the PR body and flag the pin. + +## Done when + +The PR is merged (or open with green CI and auto-merge armed), local main is synced, +and the milestone is logged to the story. diff --git a/README.md b/README.md index 53e12b58..270e7aaa 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ All modules live under `traust_core.v1` (see [Versioning](#versioning)). | **Artifact publishing** | `services.artifact_publishing` | `ctx.artifact_publisher(specs).publish(name, subject, raw, run_id)` (LLM) or `.publish_artifact(artifact, subject, run_id)` (code): exact gate → files (always) → an `artifact` row (when `database.url` is set) | writing `analysis-results` files by hand | | **Analysis results** | `repositories.objects.analysis_results` | `ctx.analysis_results().read(TriageArtifact, subject)` / `.read_all(TriageArtifact, tree)`; raw `put/get/find(subject, kind)`; today's naming convention in one place | `json.load` + `"-security-audit.json"` path building | | **Table repositories** | `repositories.traust.*`, `repositories.traust_ops`, `repositories.traust_ledger`, `repositories.contract` | one repository per contract table (`uow.findings.find_by_fingerprint(...)`, `uow.run_orders.claim(...)`); `SqlTraustUnitOfWork(engine)` opens all 31 on one connection — one transaction across sections | `sqlite3.connect`, inline SQL, per-script tables | +| **Repository observation** | `providers.forge`, `services.observation`, `services.portfolio`, `repositories.observation` | gh/glab measurements through `ProcessRunner` (`CommandSourceProvider`, strict reply validation); targets and open counts read the contract tables (`repository`, `assessment`, the `finding_status` view); `RepoStateQuery.observed` derives exposure / risk tier / freshness. Observations persist in memory until contracts lands `repository_observation` | per-script forge scraping, findings.db direct reads | | **Records (pattern)** | `repositories.sql`, `repositories.tables` | `TableSpec` (columns once) + `TableFactory` (Postgres schemas or SQLite `ops_`/`ledger_` prefixes, by engine) + `SqlTableRepository` (`add`/`get`/`update`) on SQLAlchemy Core; sqlite or Postgres by URL. `record_columns()` (mutable: UUIDv4 id + `created_at` + `updated_at`) / `append_only_columns()` (UUIDv4 id + `created_at`, no `updated_at`) + `_insert_record`/`_update_record`, database-generated via `RETURNING`; `updated_at` kept by the contracts touch trigger, SQLite `uuid()` registered by `create_database_engine` | `sqlite3.connect`, inline SQL, client-minted ids/timestamps | | **Object store** | `repositories.objects.store` | `ObjectStore.put/get/list` raw bytes by key, digest-verified (local dir; S3 later) | `write_text`, `json.dump` | | **Rendering** | `rendering` | `Report` → `MarkdownRenderer` → `publish()`; `markdown_table()` | 18 copies of `render_md` | diff --git a/src/traust_core/v1/context/config.py b/src/traust_core/v1/context/config.py index c009cf75..d07658ee 100644 --- a/src/traust_core/v1/context/config.py +++ b/src/traust_core/v1/context/config.py @@ -8,6 +8,7 @@ import yaml from traust_core.v1.errors import ConfigError +from traust_core.v1.models.inventory import DEFAULT_FORGE_HOSTS class _Model(pydantic.BaseModel): @@ -60,6 +61,7 @@ class Config(_Model): artifacts: ArtifactsConfig = ArtifactsConfig() process: ProcessConfig = ProcessConfig() allowed_repo_hosts: tuple[str, ...] = () + forge_hosts: dict[str, str] = pydantic.Field(default_factory=lambda: dict(DEFAULT_FORGE_HOSTS)) @classmethod def from_file(cls, path: Path) -> Config: diff --git a/src/traust_core/v1/context/context.py b/src/traust_core/v1/context/context.py index 9bab856f..6f17f179 100644 --- a/src/traust_core/v1/context/context.py +++ b/src/traust_core/v1/context/context.py @@ -1,6 +1,8 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass from functools import cached_property from pathlib import Path from typing import Any, TypeVar @@ -8,6 +10,7 @@ from sqlalchemy.engine import Engine from traust_core.v1.context.config import Config +from traust_core.v1.context.observation import command_source from traust_core.v1.models.artifacts import ARTIFACT_FOR_KIND from traust_core.v1.providers.base import ProviderRegistry from traust_core.v1.providers.clock import Clock, SystemClock @@ -22,7 +25,9 @@ create_database_engine, ) from traust_core.v1.repositories.contract import SqlTraustUnitOfWork +from traust_core.v1.repositories.designations import FileDesignationRepository from traust_core.v1.repositories.objects.analysis_results import AnalysisResultsRepository +from traust_core.v1.repositories.observation import InMemoryObservationStore from traust_core.v1.repositories.sql import Dialect from traust_core.v1.repositories.tables import TableFactory from traust_core.v1.services.artifact_publishing import ( @@ -30,7 +35,13 @@ PublishSpecs, SqlArtifactIndex, ) +from traust_core.v1.services.assessment import AssessmentService from traust_core.v1.services.operations import Materializer +from traust_core.v1.services.portfolio import ( + PortfolioPolicy, + PortfolioService, + RepoStateQuery, +) U = TypeVar("U", bound=UnitOfWork) @@ -106,3 +117,51 @@ def services(self) -> Services: def analysis_results(self) -> AnalysisResultsRepository: return AnalysisResultsRepository(self.object_store, ARTIFACT_FOR_KIND) + + +@dataclass(frozen=True) +class _Observation: + """What the observation door yields: the write path, the composite read, the store.""" + + service: PortfolioService + query: RepoStateQuery + store: InMemoryObservationStore + + @contextmanager + def repository_observation( + self, + *, + designations: Path | None = None, + ) -> Iterator[_Observation]: + """Observe repositories: forge facts in, derived routing state out. + + Reads inventory, assessments and open counts from the contract tables; + the observation store is in memory until contracts lands + repository_observation (NEED marker in repositories/observation.py). + """ + uow = SqlTraustUnitOfWork(self.database) + store = InMemoryObservationStore() + policy = PortfolioPolicy() + assessment = AssessmentService(uow, clock=self.clock, hosts=self.config.forge_hosts) + service = PortfolioService( + uow, + policy, + store=store, + assessment=assessment, + sources=command_source(self.config.forge_hosts, clock=self.clock), + clock=self.clock, + hosts=self.config.forge_hosts, + ) + query = RepoStateQuery( + service, + assessment, + None, # findings service: open counts come from the finding_status view + uow.open_counts, + policy, + designations=( + FileDesignationRepository(designations) if designations is not None else None + ), + clock=self.clock, + hosts=self.config.forge_hosts, + ) + yield _Observation(service=service, query=query, store=store) diff --git a/src/traust_core/v1/context/observation.py b/src/traust_core/v1/context/observation.py new file mode 100644 index 00000000..1cd6c139 --- /dev/null +++ b/src/traust_core/v1/context/observation.py @@ -0,0 +1,34 @@ +from collections.abc import Mapping + +from traust_core.v1.providers.clock import Clock, SystemClock +from traust_core.v1.providers.forge import CommandSourceProvider +from traust_core.v1.providers.process import ProcessRunner + +# Preserve forge CLI credentials during the existing batch command's migration. +_FORGE_ENV = ( + "PATH", + "HOME", + "LANG", + "LC_ALL", + "TMPDIR", + "GH_TOKEN", + "GITHUB_TOKEN", + "GH_ENTERPRISE_TOKEN", + "GITHUB_ENTERPRISE_TOKEN", + "GH_HOST", + "GH_CONFIG_DIR", + "GITLAB_TOKEN", + "GITLAB_HOST", + "GLAB_CONFIG_DIR", + "XDG_CONFIG_HOME", +) + + +def command_source( + hosts: Mapping[str, str], *, timeout_seconds: float = 120, clock: Clock | None = None +) -> CommandSourceProvider: + return CommandSourceProvider( + ProcessRunner(("gh", "glab"), keep_env=_FORGE_ENV, timeout_seconds=timeout_seconds), + clock or SystemClock(), + hosts, + ) diff --git a/src/traust_core/v1/models/assessment.py b/src/traust_core/v1/models/assessment.py index 1544a901..b0b564cb 100644 --- a/src/traust_core/v1/models/assessment.py +++ b/src/traust_core/v1/models/assessment.py @@ -129,12 +129,14 @@ class IsolationResult(Model): class Freshness(Model): - repository_ref_id: UUID + repository_ref_id: UUID | None = None last_assessed_at: datetime | None = None churn_ratio: float | None = None def age_days(self, now: datetime) -> int | None: - raise NotImplementedError("stub: replaces build_rescan_worklist audit_age_days") + if self.last_assessed_at is None: + return None + return (now.date() - self.last_assessed_at.date()).days class Population(Model): diff --git a/src/traust_core/v1/models/inventory.py b/src/traust_core/v1/models/inventory.py index a228fa72..7f5e2e8a 100644 --- a/src/traust_core/v1/models/inventory.py +++ b/src/traust_core/v1/models/inventory.py @@ -78,6 +78,9 @@ class RiskTier(StrEnum): P3 = "P3" +DEFAULT_FORGE_HOSTS: dict[str, str] = {"github.com": "github", "gitlab.com": "gitlab"} + + class ForgeHost(StrEnum): GITHUB = "github" GITLAB = "gitlab" diff --git a/src/traust_core/v1/models/observation.py b/src/traust_core/v1/models/observation.py new file mode 100644 index 00000000..735b0514 --- /dev/null +++ b/src/traust_core/v1/models/observation.py @@ -0,0 +1,118 @@ +"""Repository observation: forge measurements and the state derived from them. + +Keyed by repository_id, matching the inventory tables. +""" + +from __future__ import annotations + +from enum import StrEnum +from uuid import UUID + +from pydantic import AwareDatetime, field_serializer + +from traust_core.v1.models.base import Model +from traust_core.v1.models.inventory import ExposureClass, RiskTier +from traust_core.v1.models.values import HttpsRepoUrl + + +class Designation(StrEnum): + """Adopter-configured exposure hint; customer-reach derives from component -> product.""" + + EXTERNAL = "external" + INTERNAL_TOOLING = "internal-tooling" + + +class ComparisonStatus(StrEnum): + OK = "ok" + QUOTA_DEFERRED = "quota-deferred" + NO_PINNED_SHA = "no-pinned-sha" + UNSUPPORTED_HOST = "unsupported-host" + NO_REPO_URL = "no-repo-url" + NO_CREDENTIALS = "no-credentials" + UNREACHABLE = "unreachable" + ERROR = "error" + BAN_SUSPECTED = "ban-suspected" + NETWORK_SKIPPED = "network-skipped" + NEVER_AUDITED = "never-audited" + + +class ObservationTarget(Model): + """What the observation measures: a repository, its latest audit, its pinned commit.""" + + repository_id: UUID + repo_url: str | None + has_audit: bool = True + audit_date: str | None = None + pinned_sha: str | None = None + # NEED(artifacts): from the assessment's artifact document once that lane lands. + lines_reviewed: int | None = None + designation: Designation | None = None + repository_ref_id: UUID | None = None + + +class ForgeFacts(Model): + repository_id: UUID + repo: HttpsRepoUrl | None + repo_url: str | None + audited_sha: str | None + audit_date: str | None + has_audit: bool + status: ComparisonStatus + observed_at: AwareDatetime + pushed_at: str | None = None + archived: bool | None = None + default_branch: str | None = None + visibility: str | None = None + fork: bool | None = None + parent: str | None = None + ahead_by: int | None = None + changed_lines: int | None = None + sensitive_changed: bool | None = None + sensitive_lines: int | None = None + deps_only: bool | None = None + compare_truncated: bool | None = None + error: str | None = None + + @field_serializer("repo", when_used="json") + def serialize_repo(self, repo: HttpsRepoUrl | None) -> str | None: + return str(repo) if repo else None + + +class ChangeMetrics(Model): + changed_lines: int + sensitive_changed: bool + sensitive_lines: int + deps_only: bool + first_party_files: int + iac_files: int + threat_model_files: int + + +class DesignationRecord(Model): + match: str + designation: Designation + + +class ObservedRepoState(Model): + """The routing state derived from one repository's latest observation.""" + + repository_id: UUID + repo: HttpsRepoUrl | None + status: ComparisonStatus + risk_tier: RiskTier + exposure: ExposureClass + audit_age_days: int | None + changed_lines: int | None + churn_ratio: float | None + sensitive_changed: bool | None + sensitive_lines: int | None + deps_only: bool | None + ahead_by: int | None + compare_truncated: bool | None + push_changed: bool | None + observed_at: AwareDatetime + error: str | None = None + + @field_serializer("repo", when_used="json") + def serialize_repo(self, repo: HttpsRepoUrl | None) -> str | None: + return str(repo) if repo else None diff --git a/src/traust_core/v1/providers/forge.py b/src/traust_core/v1/providers/forge.py new file mode 100644 index 00000000..63df62b8 --- /dev/null +++ b/src/traust_core/v1/providers/forge.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import json +import re +import shutil +from collections.abc import Mapping, Sequence +from typing import Any +from urllib.parse import quote + +from traust_core.v1.errors import ServiceError, ValidationError +from traust_core.v1.models.observation import ComparisonStatus +from traust_core.v1.models.values import HttpsRepoUrl +from traust_core.v1.providers.base import Acquisition, Readiness, Result +from traust_core.v1.providers.clock import Clock +from traust_core.v1.providers.process import ProcessRunner +from traust_core.v1.providers.sources import ( + ChangedFile, + CheckoutRequest, + Comparison, + Quota, + RepoInfo, + SourceError, + Workspace, +) + +GH_COMPARE_FILE_CAP = 300 +GITLAB_DIFF_CAP = 1000 +TRIPWIRE_PATCH_CAP = 2_000_000 + + +def _failure(stderr: str) -> ComparisonStatus: + s = stderr.lower() + if any(word in s for word in ("401", "unauthorized", "authentication")): + return ComparisonStatus.NO_CREDENTIALS + if any( + word in s + for word in ( + "dial tcp", + "no such host", + "connection refused", + "i/o timeout", + "could not resolve", + "network is unreachable", + "tls handshake timeout", + ) + ): + return ComparisonStatus.UNREACHABLE + return ComparisonStatus.ERROR + + +def _text(value: bytes | str) -> str: + return value.decode(errors="replace") if isinstance(value, bytes) else value + + +class CommandSourceProvider: + name = "forge-cli" + + def __init__(self, process: ProcessRunner, clock: Clock, hosts: Mapping[str, str]) -> None: + self._process, self._clock, self._hosts = process, clock, dict(hosts) + + def check(self) -> Readiness: + missing = [binary for binary in ("gh", "glab") if shutil.which(binary) is None] + return Readiness(ready=not missing, detail=", ".join(missing)) + + def _kind(self, repo: HttpsRepoUrl) -> str: + checked = HttpsRepoUrl.parse(str(repo), self._hosts) + kind = self._hosts[checked.host] + if kind == "github" and len(checked.path.split("/")) != 2: + raise ValidationError("GitHub source requires an owner/repository URL") + if kind not in ("github", "gitlab"): + raise ValidationError(f"unsupported forge kind: {kind}") + return kind + + def _run(self, argv: Sequence[str]) -> str: + try: + result = self._process.run(argv, check=False) + except ServiceError as exc: + status = ComparisonStatus.UNREACHABLE + if argv[0] == "glab" and isinstance(exc.__cause__, FileNotFoundError): + status = ComparisonStatus.NO_CREDENTIALS + raise SourceError(str(exc)[:120], status) from exc + if result.returncode: + stderr = _text(result.stderr) + raise SourceError(stderr.strip()[:120] or "api-error", _failure(stderr)) + return _text(result.stdout) + + def _payload(self, argv: Sequence[str]) -> Mapping[str, Any]: + try: + value = json.loads(self._run(argv)) + if not isinstance(value, dict): + raise TypeError("expected a JSON object") + return value + except (ValueError, TypeError) as exc: + raise SourceError("malformed forge response", ComparisonStatus.ERROR) from exc + + def _result(self, data: Any, repo: HttpsRepoUrl) -> Result[Any]: + return Result( + data, Acquisition(provider=self.name, acquired_at=self._clock.now(), source=str(repo)) + ) + + def info(self, repo: HttpsRepoUrl) -> Result[RepoInfo]: + kind = self._kind(repo) + if kind == "github": + payload = self._payload( + [ + "gh", + "api", + f"repos/{repo.path}", + "--hostname", + repo.host, + "--jq", + "{pushed_at, archived, default_branch, visibility, " + "fork, parent: .parent.full_name}", + ] + ) + data = { + k: payload.get(k) + for k in ("pushed_at", "archived", "default_branch", "visibility", "fork", "parent") + } + else: + payload = self._payload( + ["glab", "api", "--hostname", repo.host, f"projects/{quote(repo.path, safe='')}"] + ) + try: + parent = (payload.get("forked_from_project") or {}).get("path_with_namespace") + except (AttributeError, TypeError) as exc: + raise SourceError("malformed forge response", ComparisonStatus.ERROR) from exc + data = dict( + pushed_at=payload.get("last_activity_at"), + archived=payload.get("archived"), + default_branch=payload.get("default_branch"), + visibility=payload.get("visibility"), + fork=bool(parent), + parent=parent, + ) + try: + info = RepoInfo.model_validate(dict(repo=repo, **data), strict=True) + except (ValueError, TypeError) as exc: + raise SourceError("malformed forge response", ComparisonStatus.ERROR) from exc + return self._result(info, repo) + + def compare(self, repo: HttpsRepoUrl, baseline: str, branch: str) -> Result[Comparison]: + kind = self._kind(repo) + # Existing audit reports contain abbreviated SHAs; do not weaken GitSha.parse(). + if not re.fullmatch(r"[0-9a-f]{7,40}", baseline): + raise SourceError("unsafe comparison baseline", ComparisonStatus.ERROR) + if kind == "github": + payload = self._payload( + [ + "gh", + "api", + f"repos/{repo.path}/compare/{baseline}...HEAD", + "--hostname", + repo.host, + "--jq", + "{ahead_by, status, files: [.files[] | {filename, changes}]}", + ] + ) + else: + endpoint = ( + f"projects/{quote(repo.path, safe='')}/repository/compare?" + f"from={baseline}&to={quote(branch or 'HEAD', safe='')}" + ) + payload = self._payload(["glab", "api", "--hostname", repo.host, endpoint]) + try: + if kind == "github": + files = tuple( + ChangedFile.model_validate(f, strict=True) for f in payload.get("files") or [] + ) + # Boolean/float/null ahead_by must not coerce into a measurement. + ahead = payload.get("ahead_by") + if isinstance(ahead, bool) or not isinstance(ahead, int): + raise ValueError("ahead_by must be an integer") + result = Comparison( + ahead_by=ahead, files=files, truncated=len(files) >= GH_COMPARE_FILE_CAP + ) + else: + diffs = payload.get("diffs") or [] + files = tuple( + ChangedFile( + filename=d.get("new_path") or d.get("old_path") or "", + changes=sum( + line.startswith(("+", "-")) and not line.startswith(("+++", "---")) + for line in (d.get("diff") or "").splitlines() + ), + ) + for d in diffs + ) + # GitLab omits diff content under collapsed / too_large: the comparison + # is incomplete, never a complete zero-line measurement. + omitted = bool(payload.get("too_large")) or any( + d.get("collapsed") or d.get("too_large") for d in diffs + ) + result = Comparison( + ahead_by=len(payload.get("commits") or []), + files=files, + truncated=omitted + or bool(payload.get("compare_timeout")) + or len(diffs) >= GITLAB_DIFF_CAP, + patch="\n".join(d.get("diff") or "" for d in diffs)[:TRIPWIRE_PATCH_CAP], + ) + except (ValueError, TypeError, AttributeError, KeyError) as exc: + raise SourceError("malformed forge response", ComparisonStatus.ERROR) from exc + return self._result(result, repo) + + def quota(self, repo: HttpsRepoUrl) -> Result[Quota]: + if self._kind(repo) != "github": + return self._result(Quota(remaining=None), repo) + try: + remaining = int( + self._run( + [ + "gh", + "api", + "rate_limit", + "--hostname", + repo.host, + "--jq", + ".resources.core.remaining", + ] + ).strip() + ) + except (SourceError, ValueError): + remaining = None + return self._result(Quota(remaining=remaining), repo) + + def checkout(self, request: CheckoutRequest) -> Result[Workspace]: + raise ServiceError("forge-cli only observes repositories; inject a checkout provider") + + def release(self, workspace: Workspace) -> None: + raise ServiceError("forge-cli does not own checkout workspaces") diff --git a/src/traust_core/v1/providers/sources.py b/src/traust_core/v1/providers/sources.py index f56908ca..f962f94b 100644 --- a/src/traust_core/v1/providers/sources.py +++ b/src/traust_core/v1/providers/sources.py @@ -3,9 +3,11 @@ from pathlib import Path from typing import Protocol +from traust_core.v1.errors import ServiceError from traust_core.v1.models.base import Dto from traust_core.v1.models.code_facts import LanguageStats from traust_core.v1.models.inventory import OrgName +from traust_core.v1.models.observation import ComparisonStatus from traust_core.v1.models.values import GitSha, HttpsRepoUrl from traust_core.v1.providers.base import Readiness, Result @@ -27,11 +29,37 @@ class RepoInfo(Dto): default_branch: str | None = None head: GitSha | None = None archived: bool | None = None + pushed_at: str | None = None + visibility: str | None = None + fork: bool | None = None + parent: str | None = None class OwnerHint(Dto): ... +class ChangedFile(Dto): + filename: str + changes: int + + +class Comparison(Dto): + ahead_by: int + files: tuple[ChangedFile, ...] + truncated: bool + patch: str | None = None + + +class Quota(Dto): + remaining: int | None + + +class SourceError(ServiceError): + def __init__(self, message: str, observation_status: ComparisonStatus) -> None: + super().__init__(message) + self.observation_status = observation_status + + class SourceProvider(Protocol): name: str @@ -48,3 +76,7 @@ def owners(self, repo: HttpsRepoUrl) -> Result[list[OwnerHint]]: ... def list_org(self, org: OrgName) -> Result[list[HttpsRepoUrl]]: ... def languages(self, repo: HttpsRepoUrl) -> Result[LanguageStats]: ... + + def compare(self, repo: HttpsRepoUrl, baseline: str, branch: str) -> Result[Comparison]: ... + + def quota(self, repo: HttpsRepoUrl) -> Result[Quota]: ... diff --git a/src/traust_core/v1/repositories/contract.py b/src/traust_core/v1/repositories/contract.py index 56decda1..434b864f 100644 --- a/src/traust_core/v1/repositories/contract.py +++ b/src/traust_core/v1/repositories/contract.py @@ -70,6 +70,7 @@ def _open_repositories(self) -> None: self.finding_external_refs = findings.SqlFindingExternalRefRepository(conn, tables) self.decisions = findings.SqlDecisionRepository(conn, tables) self.review_items = findings.SqlReviewItemRepository(conn, tables) + self.open_counts = findings.SqlOpenCounts(conn) self.run_batches = traust_ops.SqlRunBatchRepository(conn, tables) self.run_orders = traust_ops.SqlRunOrderRepository(conn, tables) self.spend_events = traust_ops.SqlSpendEventRepository(conn, tables) diff --git a/src/traust_core/v1/repositories/designations.py b/src/traust_core/v1/repositories/designations.py new file mode 100644 index 00000000..ba25015b --- /dev/null +++ b/src/traust_core/v1/repositories/designations.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Protocol + +from traust_core.v1.models.base import Dto +from traust_core.v1.models.observation import Designation, DesignationRecord + + +class DesignationRepository(Protocol): + def entries(self) -> list[DesignationRecord]: ... + + +class _Entry(Dto): + match: str | None = None + exposure: str | None = None + + +class _File(Dto): + designations: tuple[_Entry, ...] = () + + +class FileDesignationRepository: + def __init__(self, path: Path) -> None: + self._path = path + + def entries(self) -> list[DesignationRecord]: + try: + document = _File.model_validate_json(self._path.read_bytes()) + except (OSError, ValueError): + return [] + return [ + DesignationRecord(match=row.match, designation=Designation(row.exposure)) + for row in document.designations + if row.match and row.exposure in set(Designation) + ] + + +class InMemoryDesignationRepository: + def __init__(self, rows: list[DesignationRecord] | None = None) -> None: + self.rows = list(rows or []) + + def entries(self) -> list[DesignationRecord]: + return list(self.rows) diff --git a/src/traust_core/v1/repositories/observation.py b/src/traust_core/v1/repositories/observation.py new file mode 100644 index 00000000..388e50dd --- /dev/null +++ b/src/traust_core/v1/repositories/observation.py @@ -0,0 +1,33 @@ +"""Forge-observation storage. + +NEED(contracts): repository_observation — an append-only measurements table +(repository_id, observed_at, status, comparison metrics). Until it lands there is +deliberately no SQL implementation: core never invents a table (AGENTS.md rule). +The in-memory store carries the same stale-write semantics the table will have. +""" + +from __future__ import annotations + +from typing import Protocol +from uuid import UUID + +from traust_core.v1.models.observation import ForgeFacts + + +class ObservationStore(Protocol): + def save(self, facts: ForgeFacts) -> None: ... + def find(self, repository_id: UUID) -> ForgeFacts | None: ... + + +class InMemoryObservationStore: + def __init__(self) -> None: + self.rows: dict[str, ForgeFacts] = {} + + def save(self, facts: ForgeFacts) -> None: + previous = self.rows.get(facts.repository_id) + if previous is not None and previous.observed_at > facts.observed_at: + return # a newer observation already exists; never move time backwards + self.rows[facts.repository_id] = facts + + def find(self, repository_id: UUID) -> ForgeFacts | None: + return self.rows.get(repository_id) diff --git a/src/traust_core/v1/repositories/traust/findings/__init__.py b/src/traust_core/v1/repositories/traust/findings/__init__.py index 72a63ebc..98758623 100644 --- a/src/traust_core/v1/repositories/traust/findings/__init__.py +++ b/src/traust_core/v1/repositories/traust/findings/__init__.py @@ -35,6 +35,7 @@ ReviewItemRepository, SqlReviewItemRepository, ) +from traust_core.v1.repositories.traust.findings.status import OpenCounts, SqlOpenCounts from traust_core.v1.repositories.unit_of_work import UnitOfWork TABLES = ( @@ -56,6 +57,7 @@ class FindingsUnitOfWork(UnitOfWork, Protocol): finding_external_refs: FindingExternalRefRepository decisions: DecisionRepository review_items: ReviewItemRepository + open_counts: OpenCounts __all__ = [ @@ -67,6 +69,7 @@ class FindingsUnitOfWork(UnitOfWork, Protocol): "FindingLabelRepository", "FindingRepository", "FindingsUnitOfWork", + "OpenCounts", "ReviewItemRepository", "SqlCweRepository", "SqlDecisionRepository", @@ -74,5 +77,6 @@ class FindingsUnitOfWork(UnitOfWork, Protocol): "SqlFindingExternalRefRepository", "SqlFindingLabelRepository", "SqlFindingRepository", + "SqlOpenCounts", "SqlReviewItemRepository", ] diff --git a/src/traust_core/v1/repositories/traust/findings/status.py b/src/traust_core/v1/repositories/traust/findings/status.py new file mode 100644 index 00000000..09a8b699 --- /dev/null +++ b/src/traust_core/v1/repositories/traust/findings/status.py @@ -0,0 +1,47 @@ +"""Open counts over the finding_status view: the one definition of "open". + +The view is part of the contracts DDL (findings domain); severity is the latest +re-rating, is_open encodes the closed-set exclusions. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Protocol +from uuid import UUID + +import sqlalchemy as sa + +from traust_core.v1.models.vocabulary import Severity +from traust_core.v1.repositories.sql import SqlRepository +from traust_core.v1.repositories.tables import TRAUST, Dialect, RecordUuid + + +def _view(dialect: Dialect) -> sa.TableClause: + schema, name = TRAUST.locate("finding_status", dialect) + return sa.table( + name, + sa.column("repository_id", RecordUuid), + sa.column("severity", sa.Text), + sa.column("is_open", sa.Boolean), + schema=schema, + ) + + +class OpenCounts(Protocol): + def open_count(self, repository_id: UUID, severities: Sequence[Severity]) -> int: ... + + +class SqlOpenCounts(SqlRepository): + def open_count(self, repository_id: UUID, severities: Sequence[Severity]) -> int: + view = _view(Dialect.of(self._connection)) + stmt = ( + sa.select(sa.func.count()) + .select_from(view) + .where( + view.c.repository_id == repository_id, + view.c.is_open, + view.c.severity.in_([str(s) for s in severities]), + ) + ) + return int(self._execute(f"count open findings for {repository_id}", stmt).scalar_one()) diff --git a/src/traust_core/v1/services/assessment.py b/src/traust_core/v1/services/assessment.py index e3ee9ce5..52f7cdd3 100644 --- a/src/traust_core/v1/services/assessment.py +++ b/src/traust_core/v1/services/assessment.py @@ -1,9 +1,10 @@ from __future__ import annotations -from collections.abc import Sequence -from datetime import datetime +from collections.abc import Mapping, Sequence +from datetime import date, datetime, time from uuid import UUID +from traust_core.v1.errors import NotFoundError, ValidationError from traust_core.v1.models.artifacts import AssessmentArtifact from traust_core.v1.models.assessment import ( Assessment, @@ -20,14 +21,26 @@ ToolRun, ) from traust_core.v1.models.common import Page, PageRequest, Provenance +from traust_core.v1.models.inventory import DEFAULT_FORGE_HOSTS +from traust_core.v1.models.observation import ObservationTarget +from traust_core.v1.providers.clock import Clock, SystemClock from traust_core.v1.repositories.traust.assessments import AssessmentsUnitOfWork from traust_core.v1.services.findings.intake import FindingIntake class AssessmentService: - def __init__(self, uow: AssessmentsUnitOfWork, intake: FindingIntake) -> None: + def __init__( + self, + uow: AssessmentsUnitOfWork, + intake: FindingIntake | None = None, # record doors only + *, + clock: Clock | None = None, + hosts: Mapping[str, str] | None = None, + ) -> None: self._uow = uow self._intake = intake + self._clock = clock or SystemClock() + self._hosts = dict(hosts or DEFAULT_FORGE_HOSTS) def run_tools(self, repository_ref_id: UUID, tools: Sequence[ToolName]) -> ToolRun: raise NotImplementedError("stub: replaces run_checkov.py, adapters _contract_bridge") @@ -56,8 +69,66 @@ def list(self, filter: AssessmentFilter, page: PageRequest) -> Page[Assessment]: def population(self, as_of: datetime, filter: AssessmentFilter) -> Population: raise NotImplementedError("stub: replaces census/build_census.py, corpus/resolver") - def freshness(self, repository_ref_id: UUID) -> Freshness: - raise NotImplementedError("stub: replaces build_rescan_worklist audit_age / churn") + def freshness( + self, + repository_ref_id: UUID | None = None, + *, + baseline: ObservationTarget | None = None, + changed_lines: int | None = None, + as_of: datetime | None = None, + ) -> Freshness: + if baseline is None: + if repository_ref_id is None: + raise ValidationError( + "freshness requires a repository ref or an observation baseline" + ) + # NEED(spine): read the latest assessment row once intake populates it. + raise NotImplementedError("stub: replaces corpus/resolver.py baseline lookup") + now = as_of or self._clock.now() + if now.tzinfo is None or now.utcoffset() is None: + raise ValidationError("freshness requires a timezone-aware clock") + audited = parse_audit_date(baseline.audit_date) if baseline.has_audit else None + return Freshness( + repository_ref_id=repository_ref_id or baseline.repository_ref_id, + last_assessed_at=datetime.combine(audited, time()) if audited else None, + churn_ratio=changed_lines / baseline.lines_reviewed + if changed_lines is not None and baseline.lines_reviewed + else None, + ) + + def observation_target( + self, *, repository_id: UUID | None = None, repo_url: str | None = None + ) -> ObservationTarget: + """Resolve what an observation measures: the repository, its latest audit, + and the commit that audit pinned (the ref it occurred on).""" + if (repository_id is None) == (repo_url is None): + raise ValidationError("supply exactly one repository id or URL") + with self._uow as uow: + repo = ( + uow.repositories.get(repository_id) + if repository_id is not None + else uow.repositories.find_by_url(repo_url) + ) + if repo is None: + raise NotFoundError(f"repository not in inventory: {repository_id or repo_url}") + latest = None + for ref in uow.repository_refs.list_for(repo.id): + found = uow.assessments.latest(ref.id, AssessmentKind.SECURE_CODE) + if found is not None and ( + latest is None or found.occurred_at > latest[1].occurred_at + ): + latest = (ref, found) + if latest is None: + return ObservationTarget(repository_id=repo.id, repo_url=repo.url, has_audit=False) + ref, audit = latest + return ObservationTarget( + repository_id=repo.id, + repo_url=repo.url, + has_audit=True, + audit_date=audit.occurred_at.date().isoformat(), + pinned_sha=ref.sha, + repository_ref_id=ref.id, + ) def threats(self, filter: ThreatFilter, page: PageRequest) -> Page[Threat]: raise NotImplementedError("stub: replaces threat_current, threat-register parse_model") @@ -67,3 +138,12 @@ def coverage(self, filter: AssessmentFilter) -> list[CoverageRow]: def privileges(self, filter: AssessmentFilter) -> list[PrivilegeProfile]: raise NotImplementedError("stub: replaces operator_privilege, load_least_priv") + + +def parse_audit_date(value: str | None) -> date | None: + if not value: + return None + try: + return date.fromisoformat(value.strip()[:10]) + except ValueError: + return None diff --git a/src/traust_core/v1/services/observation.py b/src/traust_core/v1/services/observation.py new file mode 100644 index 00000000..f18b1784 --- /dev/null +++ b/src/traust_core/v1/services/observation.py @@ -0,0 +1,244 @@ +"""Forge measurement: observe a repository, classify what changed. + +Pure machinery behind PortfolioService.record_forge_facts - no persistence here. +""" + +from __future__ import annotations + +import re +from collections.abc import Callable, Mapping, Sequence +from datetime import datetime +from typing import TypeVar + +from traust_core.v1.errors import ConfigError, ValidationError +from traust_core.v1.models.observation import ( + ChangeMetrics, + ComparisonStatus, + ForgeFacts, + ObservationTarget, +) +from traust_core.v1.models.values import HttpsRepoUrl +from traust_core.v1.providers.base import Result +from traust_core.v1.providers.sources import ( + ChangedFile, + Comparison, + Quota, + RepoInfo, + SourceError, + SourceProvider, +) +from traust_core.v1.services.assessment import parse_audit_date + +T = TypeVar("T") + + +SENSITIVE_RX = re.compile( + r"auth|login|token|secret|cred|crypt|tls|ssl|cert|session|passw|" + r"rbac|scc|privile|sanitiz|escap|deserial|unmarshal|jwt|oauth|" + r"saml|acl", + re.I, +) + +DEPS_MANIFESTS = { + "go.mod", + "go.sum", + "package.json", + "package-lock.json", + "yarn.lock", + "poetry.lock", + "pom.xml", + "Cargo.toml", + "Cargo.lock", + "Gemfile", + "Gemfile.lock", +} + +TM_SURFACE_RX = re.compile( + r"(^|/)(cmd|api|apis|handler|handlers|route|routes|controller|" + r"controllers|webhook|webhooks|middleware|admission|server|" + r"endpoints?)(/|$)" + r"|(^|/)main\.(go|py|rs|java|ts|js)$" + r"|(^|/)(openapi|swagger)[^/]*\.(ya?ml|json)$" + r"|\.proto$", + re.I, +) + +BAN_STREAK_THRESHOLD = 50 + +IAC_RX = re.compile( + r"\.tf$|\.tfvars$|\.tf\.json$" + r"|(^|/)(cloudformation|cfn)/[^ ]*\.(ya?ml|json)$" + r"|\.bicep$|(^|/)azuredeploy[^/]*\.json$" +) + + +_EXCLUDE_DIR_PARTS = {"vendor", "node_modules", "third_party", "dist"} + + +class BanBreaker: + def __init__(self, threshold: int = BAN_STREAK_THRESHOLD) -> None: + self.threshold, self.streak, self.tripped = threshold, 0, False + + def record(self, ok: bool) -> None: + self.streak = 0 if ok else self.streak + 1 + if self.streak >= self.threshold: + self.tripped = True + + +def change_metrics(files: Sequence[ChangedFile]) -> ChangeMetrics: + changed, sensitive_lines, first_party, iac, threat = 0, 0, 0, 0, 0 + sensitive, deps_only = False, bool(files) + for file in files: + path = file.filename + parts, name = path.split("/"), path.rsplit("/", 1)[-1] + if not ( + name in DEPS_MANIFESTS or (name.startswith("requirements") and name.endswith(".txt")) + ): + deps_only = False + if ( + any(p in _EXCLUDE_DIR_PARTS or "_generated" in p for p in parts) + or name.endswith(".pb.go") + or name.lower().endswith(".md") + ): + continue + changed += file.changes + first_party += 1 + if SENSITIVE_RX.search(path): + sensitive = True + sensitive_lines += file.changes + iac += bool(IAC_RX.search(path)) + threat += bool(TM_SURFACE_RX.search(path)) + return ChangeMetrics( + changed_lines=changed, + sensitive_changed=sensitive, + sensitive_lines=sensitive_lines, + deps_only=deps_only, + first_party_files=first_party, + iac_files=iac, + threat_model_files=threat, + ) + + +def observation_url(raw: str | None, hosts: Mapping[str, str]) -> HttpsRepoUrl | None: + if raw is None: + return None + try: + typed = HttpsRepoUrl.parse(raw, hosts) + except ValueError: + return None + if hosts[typed.host] == "github": + return HttpsRepoUrl(typed.host, "/".join(typed.path.split("/")[:2]).lower()) + return typed + + +T = TypeVar("T") + + +def _fetch(fetch: Callable[[], Result[T]], expected: type[T]) -> T: + try: + data = fetch().data + if not isinstance(data, expected): + raise TypeError("unexpected forge result type") + return data + except SourceError: + raise + except OSError as exc: + raise SourceError(str(exc), ComparisonStatus.UNREACHABLE) from exc + except (AttributeError, KeyError, TypeError, ValueError) as exc: + raise SourceError( + f"malformed forge response ({type(exc).__name__})", ComparisonStatus.ERROR + ) from exc + + +def observe_forge_facts( + target: ObservationTarget, + sources: SourceProvider | None, + now: datetime, + *, + hosts: Mapping[str, str], + network: bool = True, + breaker: BanBreaker | None = None, + quota_deferred: bool | None = None, +) -> ForgeFacts: + if now.tzinfo is None or now.utcoffset() is None: + raise ValidationError("observation time must include a timezone") + repo = observation_url(target.repo_url, hosts) + status = ComparisonStatus.OK + if target.repo_url is None: + status = ComparisonStatus.NO_REPO_URL + elif repo is None: + status = ComparisonStatus.UNSUPPORTED_HOST + elif not target.has_audit: + status = ComparisonStatus.NEVER_AUDITED + elif not network: + status = ComparisonStatus.NETWORK_SKIPPED + facts = ForgeFacts( + repository_id=target.repository_id, + repo=repo, + repo_url=target.repo_url, + audited_sha=target.pinned_sha, + audit_date=target.audit_date, + has_audit=target.has_audit, + status=status, + observed_at=now, + ) + if status != ComparisonStatus.OK: + return facts + if sources is None or repo is None: + raise ConfigError("repository observation requires a source provider") + github = hosts[repo.host] == "github" + breaker = breaker if breaker is not None else BanBreaker() + if github and breaker.tripped: + return facts.model_copy( + update={ + "status": ComparisonStatus.BAN_SUSPECTED, + "error": "skipped: consecutive-error streak suggests a GitHub secondary ban", + } + ) + try: + info = _fetch(lambda: sources.info(repo), RepoInfo) + if info.repo != repo: + raise SourceError( + "forge response identifies a different repository", ComparisonStatus.ERROR + ) + except SourceError as exc: + if github: + breaker.record(False) + return facts.model_copy(update={"status": exc.observation_status, "error": str(exc)}) + if github: + breaker.record(True) + facts = facts.model_copy( + update={ + key: getattr(info, key) + for key in ("pushed_at", "archived", "default_branch", "visibility", "fork", "parent") + } + ) + pushed, audited = parse_audit_date(info.pushed_at), parse_audit_date(target.audit_date) + if not (pushed and audited and pushed >= audited): + return facts + if not target.pinned_sha: + return facts.model_copy(update={"status": ComparisonStatus.NO_PINNED_SHA}) + try: + deferred = quota_deferred + if deferred is None: + quota = _fetch(lambda: sources.quota(repo), Quota) if github else Quota(remaining=None) + deferred = quota.remaining is not None and quota.remaining <= 0 + if deferred: + return facts.model_copy(update={"status": ComparisonStatus.QUOTA_DEFERRED}) + compared = _fetch( + lambda: sources.compare(repo, target.pinned_sha, info.default_branch or "HEAD"), + Comparison, + ) + except SourceError as exc: + return facts.model_copy(update={"status": exc.observation_status, "error": str(exc)}) + metrics = change_metrics(compared.files) + return facts.model_copy( + update=dict( + ahead_by=compared.ahead_by, + compare_truncated=compared.truncated, + changed_lines=metrics.changed_lines, + sensitive_changed=metrics.sensitive_changed, + sensitive_lines=metrics.sensitive_lines, + deps_only=metrics.deps_only, + ) + ) diff --git a/src/traust_core/v1/services/portfolio.py b/src/traust_core/v1/services/portfolio.py index a031a02a..6e55a8dd 100644 --- a/src/traust_core/v1/services/portfolio.py +++ b/src/traust_core/v1/services/portfolio.py @@ -1,10 +1,13 @@ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime, timedelta from uuid import UUID +from traust_core.v1.errors import ConfigError, NotFoundError, ValidationError from traust_core.v1.models.common import Page, PageRequest from traust_core.v1.models.inventory import ( + DEFAULT_FORGE_HOSTS, Component, ComponentFilter, ExposureClass, @@ -26,17 +29,106 @@ SyncReport, VersionLabel, ) +from traust_core.v1.models.observation import ( + Designation, + ForgeFacts, + ObservationTarget, + ObservedRepoState, +) from traust_core.v1.models.values import GitSha, HttpsRepoUrl -from traust_core.v1.models.vocabulary import RefKind +from traust_core.v1.models.vocabulary import RefKind, Severity +from traust_core.v1.providers.clock import Clock, SystemClock from traust_core.v1.providers.sources import SourceProvider +from traust_core.v1.repositories.designations import DesignationRepository +from traust_core.v1.repositories.observation import ObservationStore +from traust_core.v1.repositories.traust.findings import OpenCounts from traust_core.v1.repositories.traust.inventory import InventoryUnitOfWork -from traust_core.v1.services.assessment import AssessmentService +from traust_core.v1.services.assessment import AssessmentService, parse_audit_date from traust_core.v1.services.findings.service import FindingService +from traust_core.v1.services.observation import BanBreaker, observe_forge_facts + +DORMANT_DAYS = 365 + +P0_MIN_LIVE = 5 + + +class PortfolioPolicy: + def __init__(self, *, p0_min_live: int = P0_MIN_LIVE, dormant_days: int = DORMANT_DAYS) -> None: + if p0_min_live < 1 or dormant_days < 0: + raise ValidationError("portfolio policy thresholds must be positive") + self.p0_min_live, self.dormant_days = p0_min_live, dormant_days + + def exposure_class( + self, designation: Designation | None, public: bool | None, public_upstream: bool | None + ) -> ExposureClass: + if public: + return ( + ExposureClass.PUBLIC_INTERNAL + if designation == Designation.INTERNAL_TOOLING + else ExposureClass.PUBLIC_EXTERNAL + ) + if designation == Designation.EXTERNAL: + return ( + ExposureClass.PUBLIC_EXTERNAL if public_upstream else ExposureClass.PRIVATE_EXTERNAL + ) + return ExposureClass.PUBLIC_INTERNAL if public_upstream else ExposureClass.PRIVATE_INTERNAL + + def risk_tier(self, open_critical_high: int, archived: bool, dormant: bool) -> RiskTier: + if open_critical_high >= self.p0_min_live: + return RiskTier.P0 + if open_critical_high >= 1: + return RiskTier.P1 + return RiskTier.P3 if archived or dormant else RiskTier.P2 class PortfolioService: - def __init__(self, uow: InventoryUnitOfWork) -> None: - self._uow = uow + def __init__( + self, + uow: InventoryUnitOfWork, + policy: PortfolioPolicy, + *, + store: ObservationStore, + assessment: AssessmentService | None = None, + sources: SourceProvider | None = None, + clock: Clock | None = None, + hosts: Mapping[str, str] | None = None, + ) -> None: + self._uow, self._policy, self._store = uow, policy, store + self._assessment, self._sources = assessment, sources + self._clock = clock or SystemClock() + self._hosts = dict(hosts or DEFAULT_FORGE_HOSTS) + + def _target( + self, *, repository_id: UUID | None = None, repo_url: str | None = None + ) -> ObservationTarget: + if self._assessment is None: + raise ConfigError("repository observation requires AssessmentService") + return self._assessment.observation_target(repository_id=repository_id, repo_url=repo_url) + + def record_forge_facts( + self, + *, + repository_id: UUID | None = None, + repo_url: str | None = None, + network: bool = True, + breaker: BanBreaker | None = None, + quota_deferred: bool | None = None, + ) -> ForgeFacts: + target = self._target(repository_id=repository_id, repo_url=repo_url) + facts = observe_forge_facts( + target, + self._sources, + self._clock.now(), + hosts=self._hosts, + network=network, + breaker=breaker, + quota_deferred=quota_deferred, + ) + self._store.save(facts) + return facts + + def forge_facts(self, repository_id: UUID) -> ForgeFacts | None: + return self._store.find(repository_id) def register_product(self, slug: ProductSlug, name: str) -> Product: raise NotImplementedError("stub: replaces registry/products.py, inventory.py") @@ -105,12 +197,88 @@ def import_owners(self, source: OwnershipSource) -> SyncReport: class RepoStateQuery: + """Composite read across portfolio, assessments and findings (traust#58).""" + def __init__( - self, portfolio: PortfolioService, assessments: AssessmentService, findings: FindingService + self, + portfolio: PortfolioService, + assessments: AssessmentService, + findings: FindingService, + counts: OpenCounts, + policy: PortfolioPolicy, + *, + designations: DesignationRepository | None = None, + clock: Clock | None = None, + hosts: Mapping[str, str] | None = None, ) -> None: self._portfolio = portfolio self._assessments = assessments self._findings = findings + self._counts, self._policy = counts, policy + self._designations = designations + self._clock = clock or SystemClock() + self._hosts = dict(hosts or DEFAULT_FORGE_HOSTS) + + def observed( + self, *, repository_id: UUID | None = None, repo_url: str | None = None + ) -> ObservedRepoState: + """The routing state derived from one repository's latest observation.""" + target = self._assessments.observation_target( + repository_id=repository_id, repo_url=repo_url + ) + facts = self._portfolio.forge_facts(target.repository_id) + if facts is None: + raise NotFoundError(f"no forge observation for {target.repo_url}; observe it first") + if (facts.audited_sha, facts.audit_date, facts.has_audit) != ( + target.pinned_sha, + target.audit_date, + target.has_audit, + ): + raise NotFoundError(f"audit baseline changed for {target.repo_url}; observe it again") + live = self._counts.open_count(target.repository_id, (Severity.CRITICAL, Severity.HIGH)) + now = self._clock.now() + fresh = self._assessments.freshness( + baseline=target, changed_lines=facts.changed_lines, as_of=now + ) + designation = target.designation + if self._designations: + designation = next( + ( + row.designation + for row in self._designations.entries() + if target.repo_url + and ( + target.repo_url == row.match + or (row.match.endswith("/") and target.repo_url.startswith(row.match)) + ) + ), + target.designation, + ) + pushed = parse_audit_date(facts.pushed_at) + audited = parse_audit_date(target.audit_date) if target.has_audit else None + dormant = bool(pushed and (now.date() - pushed).days > self._policy.dormant_days) + return ObservedRepoState( + repository_id=target.repository_id, + repo=facts.repo, + status=facts.status, + risk_tier=self._policy.risk_tier(live, bool(facts.archived), dormant), + exposure=self._policy.exposure_class( + designation, + (facts.visibility or "").lower() == "public", + bool(facts.fork and facts.parent), + ), + audit_age_days=fresh.age_days(now), + churn_ratio=fresh.churn_ratio, + changed_lines=facts.changed_lines, + sensitive_changed=facts.sensitive_changed, + sensitive_lines=facts.sensitive_lines, + deps_only=facts.deps_only, + ahead_by=facts.ahead_by, + compare_truncated=facts.compare_truncated, + push_changed=pushed >= audited if pushed and audited else None, + observed_at=facts.observed_at, + error=facts.error, + ) def repo_state(self, repository_id: UUID) -> RepoState: raise NotImplementedError("stub: replaces traust#58 observe_repo") diff --git a/tests/fixtures/observations/recorded-github.json b/tests/fixtures/observations/recorded-github.json new file mode 100644 index 00000000..04326da4 --- /dev/null +++ b/tests/fixtures/observations/recorded-github.json @@ -0,0 +1,31 @@ +{ + "recorded_at": "2026-10-06T21:19:56.046950+00:00", + "repository_url": "https://github.com/traust-security/traust", + "base_sha": "c98dad755f10375e99ebfdfe5916e7828d4ee3c1", + "head_sha": "2ad90c14300df343a790182d837b43ebc5263162", + "provenance": "Recorded from the public GitHub REST API. Only the fields selected by the legacy gh --jq calls are retained; no headers, credentials, identities, patches, or commit messages are stored. Audit baseline fields used with this recording are synthetic test inputs, not actual audit claims.", + "info": { + "endpoint": "https://api.github.com/repos/traust-security/traust", + "response": { + "archived": false, + "default_branch": "main", + "fork": false, + "parent": null, + "pushed_at": "2026-10-06T16:55:49Z", + "visibility": "public" + } + }, + "compare": { + "endpoint": "https://api.github.com/repos/traust-security/traust/compare/c98dad755f10375e99ebfdfe5916e7828d4ee3c1...2ad90c14300df343a790182d837b43ebc5263162", + "response": { + "ahead_by": 2, + "files": [ + { + "changes": 711, + "filename": "uv.lock" + } + ], + "status": "ahead" + } + } +} diff --git a/tests/fixtures/observations/reference.json b/tests/fixtures/observations/reference.json new file mode 100644 index 00000000..13cf5402 --- /dev/null +++ b/tests/fixtures/observations/reference.json @@ -0,0 +1,727 @@ +{ + "source_revision": "2ad90c14300df343a790182d837b43ebc5263162", + "source_sha256": "742643c232656bfd240e2f280d664277d18e0ff24f949122dcee12ce3306f80f", + "now": "2026-10-06T12:00:00+00:00", + "cases": [ + { + "name": "sensitive-first-party", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": 0.024 + }, + "status": "ok", + "observed": { + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "ahead_by": 2, + "truncated": false + } + }, + { + "name": "measured-zero", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 0, + "truncated": false, + "files": [] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 0, + "S": false, + "S_lines": 0, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 0, + "R": 0.0 + }, + "status": "ok", + "observed": { + "C": 0, + "S": false, + "S_lines": 0, + "deps_only": false, + "ahead_by": 0, + "truncated": false + } + }, + { + "name": "dependencies-only", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "go.mod", + "changes": 10 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 10, + "S": false, + "S_lines": 0, + "deps_only": true, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": 0.001 + }, + "status": "ok", + "observed": { + "C": 10, + "S": false, + "S_lines": 0, + "deps_only": true, + "ahead_by": 2, + "truncated": false + } + }, + { + "name": "unchanged", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-08-01T00:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 0, + "S": false, + "S_lines": 0, + "deps_only": false, + "audit_age_days": 35, + "changed": false, + "ahead_by": null, + "R": 0.0 + }, + "status": "ok", + "observed": { + "C": null, + "S": null, + "S_lines": null, + "deps_only": null, + "ahead_by": null, + "truncated": null + } + }, + { + "name": "same-day-push", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-09-01T00:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": 0.024 + }, + "status": "ok", + "observed": { + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "ahead_by": 2, + "truncated": false + } + }, + { + "name": "missing-denominator", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": null, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": null + }, + "status": "ok", + "observed": { + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "ahead_by": 2, + "truncated": false + } + }, + { + "name": "missing-date", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": null, + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 0, + "S": false, + "S_lines": 0, + "deps_only": false, + "audit_age_days": null, + "changed": false, + "ahead_by": null, + "R": 0.0 + }, + "status": "ok", + "observed": { + "C": null, + "S": null, + "S_lines": null, + "deps_only": null, + "ahead_by": null, + "truncated": null + } + }, + { + "name": "archived-live-risk", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 5 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": true, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P0", + "exposure": "public-external", + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": 0.024 + }, + "status": "ok", + "observed": { + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "ahead_by": 2, + "truncated": false + } + }, + { + "name": "private-upstream", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "private", + "fork": true, + "parent": "o/upstream" + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-internal", + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": 0.024 + }, + "status": "ok", + "observed": { + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "ahead_by": 2, + "truncated": false + } + }, + { + "name": "info-failure", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": false, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null, + "kind": "unreachable", + "error": "offline" + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "private-internal", + "C": 0, + "S": false, + "S_lines": 0, + "deps_only": false, + "audit_age_days": 35, + "changed": false, + "ahead_by": null, + "R": 0.0 + }, + "status": "unreachable", + "observed": { + "C": null, + "S": null, + "S_lines": null, + "deps_only": null, + "ahead_by": null, + "truncated": null + } + }, + { + "name": "compare-failure", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/o/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": false, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ], + "kind": "no-credentials", + "error": "offline" + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 0, + "S": false, + "S_lines": 0, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": null, + "R": 0.0 + }, + "status": "no-credentials", + "observed": { + "C": null, + "S": null, + "S_lines": null, + "deps_only": null, + "ahead_by": null, + "truncated": null + } + }, + { + "name": "gitlab-subgroup", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://gitlab.com/g/sub/repo", + "audit_date": "2026-09-01", + "pinned_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-01T12:00:00Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": true, + "files": [ + { + "filename": "pkg/auth.go", + "changes": 240 + }, + { + "filename": "vendor/auth.go", + "changes": 9000 + } + ], + "patch": "diff" + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": 0.024 + }, + "status": "ok", + "observed": { + "C": 240, + "S": true, + "S_lines": 240, + "deps_only": false, + "ahead_by": 2, + "truncated": true + } + }, + { + "name": "recorded-public-github", + "repo": { + "repo_key": "fixture/repo", + "repo_url": "https://github.com/traust-security/traust", + "audit_date": "2026-09-01", + "pinned_sha": "c98dad755f10375e99ebfdfe5916e7828d4ee3c1", + "lines_reviewed": 10000, + "live_crit_high": 1 + }, + "info": { + "ok": true, + "pushed_at": "2026-10-06T16:55:49Z", + "archived": false, + "default_branch": "main", + "visibility": "public", + "fork": false, + "parent": null + }, + "compare": { + "ok": true, + "ahead_by": 2, + "truncated": false, + "files": [ + { + "changes": 711, + "filename": "uv.lock" + } + ] + }, + "ctx": { + "tier": "P1", + "exposure": "public-external", + "C": 711, + "S": false, + "S_lines": 0, + "deps_only": false, + "audit_age_days": 35, + "changed": true, + "ahead_by": 2, + "R": 0.0711 + }, + "status": "ok", + "observed": { + "C": 711, + "S": false, + "S_lines": 0, + "deps_only": false, + "ahead_by": 2, + "truncated": false + } + } + ], + "recorded_forge": { + "file": "recorded-github.json", + "sha256": "4271c753056cc2436c0dee0f0e7cd848023baa064c1fddfee5cea64fbeef53b4" + } +} diff --git a/tests/v1/context/test_observation.py b/tests/v1/context/test_observation.py new file mode 100644 index 00000000..bb5c430a --- /dev/null +++ b/tests/v1/context/test_observation.py @@ -0,0 +1,24 @@ +import pytest + +from tests.v1.observation_support import NOW +from traust_core.v1.context.observation import command_source +from traust_core.v1.errors import ValidationError +from traust_core.v1.models.values import HttpsRepoUrl +from traust_core.v1.providers.clock import FixedClock +from traust_core.v1.providers.process import ProcessResult, ProcessRunner + + +def test_batch_source_respects_the_supplied_host_allowlist_and_clock(monkeypatch): + calls = [] + + def run(self, argv, **kwargs): + calls.append(tuple(argv)) + return ProcessResult(tuple(argv), 0, b"{}", b"", 0) + + monkeypatch.setattr(ProcessRunner, "run", run) + source = command_source({"github.com": "github"}, clock=FixedClock(NOW)) + with pytest.raises(ValidationError): + source.info(HttpsRepoUrl("example.com", "o/r")) + assert calls == [] + assert source.info(HttpsRepoUrl("github.com", "o/r")).acquisition.acquired_at == NOW + assert len(calls) == 1 diff --git a/tests/v1/observation_support.py b/tests/v1/observation_support.py new file mode 100644 index 00000000..d1c51aec --- /dev/null +++ b/tests/v1/observation_support.py @@ -0,0 +1,143 @@ +from __future__ import annotations + +import json +from collections.abc import Sequence +from datetime import datetime +from pathlib import Path +from uuid import NAMESPACE_URL, UUID, uuid5 + +from traust_core.v1.models.observation import ComparisonStatus, ObservationTarget +from traust_core.v1.models.vocabulary import Severity +from traust_core.v1.providers.base import Acquisition, Result +from traust_core.v1.providers.clock import FixedClock +from traust_core.v1.providers.sources import ChangedFile, Comparison, Quota, RepoInfo, SourceError +from traust_core.v1.repositories.observation import InMemoryObservationStore +from traust_core.v1.services.assessment import AssessmentService +from traust_core.v1.services.portfolio import PortfolioPolicy, PortfolioService, RepoStateQuery + +REFERENCE = json.loads( + (Path(__file__).parents[1] / "fixtures/observations/reference.json").read_text() +) +NOW = datetime.fromisoformat(REFERENCE["now"]) +HOSTS = {"github.com": "github", "gitlab.com": "gitlab"} + + +def repository_id_for(repo_url: str | None, repo_key: str) -> UUID: + """A stable repository id per fixture repo (the model supplies real UUIDs).""" + return uuid5(NAMESPACE_URL, repo_url or f"key:{repo_key}") + + +class FakeSource: + def __init__(self, info=None, comparison=None, quota=100): + self.info_data = info if info is not None else REFERENCE["cases"][0]["info"] + self.comparison_data = ( + comparison if comparison is not None else REFERENCE["cases"][0]["compare"] + ) + self.remaining = quota + self.calls = [] + + def result(self, data): + return Result(data, Acquisition(provider="fake", acquired_at=NOW)) + + def info(self, repo): + self.calls.append("info") + data = {k: v for k, v in self.info_data.items() if k != "ok"} + if not self.info_data["ok"]: + raise SourceError(data.get("error") or "source failed", ComparisonStatus(data["kind"])) + return self.result(RepoInfo(repo=repo, **data)) + + def compare(self, repo, baseline, branch): + self.calls.append("compare") + data = self.comparison_data + if not data["ok"]: + raise SourceError(data.get("error") or "source failed", ComparisonStatus(data["kind"])) + return self.result( + Comparison( + ahead_by=data["ahead_by"], + truncated=data["truncated"], + files=tuple(ChangedFile(**f) for f in data["files"]), + ) + ) + + def quota(self, repo): + self.calls.append("quota") + return self.result(Quota(remaining=self.remaining)) + + +class FixedOpenCounts: + """The finding_status view double: N open critical/high findings for a repo.""" + + def __init__(self) -> None: + self.counts: dict[UUID, int] = {} + + def open_count(self, repository_id: UUID, severities: Sequence[Severity]) -> int: + if Severity.CRITICAL not in severities and Severity.HIGH not in severities: + return 0 + return self.counts.get(repository_id, 0) + + +class FixedBaselineAssessment(AssessmentService): + def __init__(self, target, clock): + super().__init__(None, clock=clock) # uow unused: target and baseline are fixed + self.target = target + + def observation_target(self, **kwargs): + return self.target + + +def observer(case=None, *, store=None, **changes): + case = case or REFERENCE["cases"][0] + data = {**case["repo"], **changes} + live = data.pop("live_crit_high", 0) + repo_key = data.pop("repo_key") + repo_url = data.pop("repo_url", None) + target = ObservationTarget( + repository_id=repository_id_for(repo_url, repo_key), + repo_url=repo_url, + **data, + ) + clock = FixedClock(NOW) + assessment = FixedBaselineAssessment(target, clock) + counts = FixedOpenCounts() + counts.counts[target.repository_id] = live + sources = FakeSource(case["info"], case["compare"]) + store = store if store is not None else InMemoryObservationStore() + service = PortfolioService( + None, # contract uow unused: observation writes go to the store + PortfolioPolicy(), + store=store, + assessment=assessment, + sources=sources, + clock=clock, + hosts=HOSTS, + ) + harness = _Harness( + service, + RepoStateQuery( + service, + assessment, + None, # findings service: open counts come from the counts port + counts, + PortfolioPolicy(), + clock=clock, + hosts=HOSTS, + ), + ) + return harness, store, assessment, counts, sources + + +class _Harness: + """Exposes both the observation write path and the composite read.""" + + def __init__(self, service, query): + self._service, self.query = service, query + + @property + def repository_id(self) -> UUID: + return self._service._assessment.target.repository_id + + def record_forge_facts(self, **kwargs): + return self._service.record_forge_facts(**kwargs) + + def observed(self, **kwargs): + return self.query.observed(**kwargs) diff --git a/tests/v1/providers/test_forge.py b/tests/v1/providers/test_forge.py new file mode 100644 index 00000000..730ca307 --- /dev/null +++ b/tests/v1/providers/test_forge.py @@ -0,0 +1,140 @@ +import json + +import pytest + +from tests.v1.observation_support import HOSTS, NOW +from traust_core.v1.models.values import HttpsRepoUrl +from traust_core.v1.providers.clock import FixedClock +from traust_core.v1.providers.forge import CommandSourceProvider +from traust_core.v1.providers.process import ProcessResult +from traust_core.v1.providers.sources import SourceError + + +class Replay: + def __init__(self, payload, returncode=0, stderr=b""): + self.payload, self.returncode, self.stderr = payload, returncode, stderr + self.calls = [] + + def run(self, argv, **kwargs): + self.calls.append(tuple(argv)) + return ProcessResult(tuple(argv), self.returncode, self.payload, self.stderr, 0) + + +def source(payload, **kwargs): + replay = Replay(payload, **kwargs) + return CommandSourceProvider(replay, FixedClock(NOW), HOSTS), replay + + +@pytest.mark.parametrize( + "host,phase,payload", + [ + ("github.com", "info", b"{bad"), + ("github.com", "info", b"[]"), + ("github.com", "info", b'{"pushed_at": 123}'), + ("github.com", "info", b'{"visibility": ["public"]}'), + ("github.com", "compare", b"null"), + ("github.com", "compare", b'{"ahead_by": "oops", "files": []}'), + ("github.com", "compare", b'{"ahead_by": null, "files": []}'), + ("github.com", "compare", b'{"ahead_by": true, "files": []}'), + ("github.com", "compare", b'{"ahead_by": 1.5, "files": []}'), + ("github.com", "compare", b'{"ahead_by": 1, "files": [{}]}'), + ("github.com", "compare", b'{"files": [{"filename": "a.py", "changes": {}}]}'), + ("gitlab.com", "info", b"[]"), + ("gitlab.com", "info", b'{"forked_from_project": ["invalid"]}'), + ("gitlab.com", "compare", b'{"diffs": [null]}'), + ("gitlab.com", "compare", b'{"diffs": [{"diff": 123}]}'), + ], +) +def test_malformed_response_is_a_typed_source_error(host, phase, payload): + provider, _ = source(payload) + repo = HttpsRepoUrl(host, "o/r") + with pytest.raises(SourceError) as error: + if phase == "info": + provider.info(repo) + else: + provider.compare(repo, "abcdef1", "main") + assert error.value.observation_status == "error" + assert payload.decode() not in str(error.value) + + +@pytest.mark.parametrize( + "stderr,status", + [ + ("HTTP 401 Unauthorized", "no-credentials"), + ("dial tcp failed", "unreachable"), + ("HTTP 500", "error"), + ], +) +def test_command_failure_classification(stderr, status): + provider, _ = source(b"", returncode=1, stderr=stderr.encode()) + with pytest.raises(SourceError) as error: + provider.info(HttpsRepoUrl("github.com", "o/r")) + assert error.value.observation_status == status + + +def test_gitlab_subgroups_branch_escaping_and_diff_metrics(): + data = { + "commits": [{}, {}], + "diffs": [{"new_path": "a.py", "diff": "--- a\n+++ b\n@@ 1 @@\n-old\n+new"}], + } + provider, replay = source(json.dumps(data).encode()) + result = provider.compare( + HttpsRepoUrl("gitlab.com", "o/sub/r"), "abcdef1", "release/x&secret=1" + ) + assert replay.calls == [ + ( + "glab", + "api", + "--hostname", + "gitlab.com", + "projects/o%2Fsub%2Fr/repository/compare?from=abcdef1&to=release%2Fx%26secret%3D1", + ) + ] + assert result.data.ahead_by == 2 and result.data.files[0].changes == 2 + assert result.acquisition.acquired_at == NOW + + +@pytest.mark.parametrize( + "data", + [ + # A collapsed entry omits its diff content entirely. + {"commits": [{"id": "a"}], "diffs": [{"new_path": "b.py", "collapsed": True}]}, + # Per-diff too_large: same omission, different flag. + {"commits": [{"id": "a"}], "diffs": [{"new_path": "b.py", "too_large": True}]}, + # Response-level too_large can accompany partial diffs. + { + "commits": [{"id": "a"}], + "too_large": True, + "diffs": [{"new_path": "a.py", "diff": "+x"}], + }, + ], + ids=["collapsed", "per-diff-too-large", "response-too-large"], +) +def test_gitlab_omitted_diffs_mark_the_comparison_incomplete(data): + provider, _ = source(json.dumps(data).encode()) + result = provider.compare(HttpsRepoUrl("gitlab.com", "o/r"), "abcdef1", "main") + assert result.data.truncated is True + + +@pytest.mark.parametrize("sha", ["bad", "-ref", "a" * 40 + "?x=y"]) +def test_invalid_baseline_executes_nothing(sha): + provider, replay = source(b"{}") + with pytest.raises(SourceError): + provider.compare(HttpsRepoUrl("github.com", "o/r"), sha, "HEAD") + assert replay.calls == [] + + +def test_undeterminable_quota_remains_unknown(): + provider, _ = source(b"bad") + assert provider.quota(HttpsRepoUrl("github.com", "o/r")).data.remaining is None + + +def test_github_commands_explicitly_select_the_validated_host(): + provider, replay = source(b'{"ahead_by": 0, "files": []}') + repo = HttpsRepoUrl("github.com", "o/r") + provider.info(repo) + provider.compare(repo, "abcdef1", "HEAD") + provider.quota(repo) + assert len(replay.calls) == 3 + for argv in replay.calls: + assert argv[argv.index("--hostname") + 1] == repo.host diff --git a/tests/v1/repositories/tables/test_observation_target.py b/tests/v1/repositories/tables/test_observation_target.py new file mode 100644 index 00000000..ba493a3a --- /dev/null +++ b/tests/v1/repositories/tables/test_observation_target.py @@ -0,0 +1,101 @@ +"""observation_target and open counts over the contract tables (SQLite + Postgres).""" + +from __future__ import annotations + +from datetime import UTC, datetime +from uuid import UUID + +import pytest +from sqlalchemy.engine import Engine + +from tests.v1.repositories.tables.conftest import Seed +from traust_core.v1.errors import NotFoundError, ValidationError +from traust_core.v1.models.assessment import AssessmentKind +from traust_core.v1.models.observation import ObservationTarget +from traust_core.v1.repositories.contract import SqlTraustUnitOfWork +from traust_core.v1.services.assessment import AssessmentService + + +def service(engine: Engine) -> AssessmentService: + """A service over its own unit of work, as the context door builds it.""" + return AssessmentService(SqlTraustUnitOfWork(engine)) + + +def test_resolves_latest_audit_and_pinned_commit(uow, seed: Seed, engine: Engine): + repo = seed.repository(url="https://github.com/org/audited") + old_ref = seed.repository_ref(repository=repo, sha="a" * 40) + new_ref = seed.repository_ref(repository=repo, sha="b" * 40) + seed.assessment( + ref=old_ref, + kind=AssessmentKind.SECURE_CODE, + occurred_at=datetime(2026, 1, 1, tzinfo=UTC), + ) + seed.assessment( + ref=new_ref, + kind=AssessmentKind.SECURE_CODE, + occurred_at=datetime(2026, 9, 1, tzinfo=UTC), + ) + # a non-audit assessment on a newer ref must not win + seed.assessment( + ref=new_ref, + kind=AssessmentKind.THREAT_MODEL, + occurred_at=datetime(2026, 10, 1, tzinfo=UTC), + ) + uow.commit() + + target = service(engine).observation_target(repo_url="https://github.com/org/audited") + assert target == ObservationTarget( + repository_id=repo.id, + repo_url="https://github.com/org/audited", + has_audit=True, + audit_date="2026-09-01", + pinned_sha="b" * 40, + repository_ref_id=new_ref.id, + ) + by_id = service(engine).observation_target(repository_id=repo.id) + assert by_id == target + + +def test_observation_target_without_an_audit_is_never_audited(uow, seed: Seed, engine: Engine): + repo = seed.repository(url="https://github.com/org/never-audited") + uow.commit() + + target = service(engine).observation_target(repo_url="https://github.com/org/never-audited") + assert target.has_audit is False + assert target.repository_id == repo.id + + +def test_observation_target_requires_inventory_membership(uow, seed: Seed, engine: Engine): + with pytest.raises(NotFoundError): + service(engine).observation_target(repo_url="https://github.com/org/unknown") + + +def test_observation_target_takes_exactly_one_selector(engine: Engine): + with pytest.raises(ValidationError): + service(engine).observation_target() + with pytest.raises(ValidationError): + service(engine).observation_target( + repository_id=UUID(int=1), repo_url="https://github.com/org/x" + ) + + +def test_open_counts_read_the_finding_status_view(uow, seed: Seed, engine: Engine): + from traust_core.v1.models.vocabulary import Severity + + repo = seed.repository() + counts = uow.open_counts + assert counts.open_count(repo.id, (Severity.CRITICAL, Severity.HIGH)) == 0 + critical, high, medium = ( + seed.finding(repository=repo), + seed.finding(repository=repo), + seed.finding(repository=repo), + ) + # severity rides on the occurrence (the finding row is identity only) + seed.occurrence(finding=critical, severity="critical") + seed.occurrence(finding=high, severity="high") + seed.occurrence(finding=medium, severity="medium") + # decisions close two of the three: one false_positive, one resolved + seed.decision(finding=critical, validity="false_positive") + seed.decision(finding=medium, resolution="resolved") + assert counts.open_count(repo.id, (Severity.CRITICAL, Severity.HIGH)) == 1 + assert counts.open_count(repo.id, ()) == 0 diff --git a/tests/v1/repositories/test_designations.py b/tests/v1/repositories/test_designations.py new file mode 100644 index 00000000..c412af13 --- /dev/null +++ b/tests/v1/repositories/test_designations.py @@ -0,0 +1,41 @@ +import json + +import pytest + +from traust_core.v1.models.observation import Designation, DesignationRecord +from traust_core.v1.repositories.designations import ( + FileDesignationRepository, + InMemoryDesignationRepository, +) + + +@pytest.mark.parametrize("backend", ["memory", "file"]) +def test_designations_preserve_order_and_return_independent_lists(backend, tmp_path): + rows = [ + DesignationRecord(match="https://github.com/o/", designation=Designation.EXTERNAL), + DesignationRecord(match="https://github.com/o/r", designation=Designation.INTERNAL_TOOLING), + ] + path = tmp_path / "designations.json" + path.write_text( + json.dumps( + {"designations": [{"match": row.match, "exposure": row.designation} for row in rows]} + ) + ) + repo = ( + InMemoryDesignationRepository(rows) + if backend == "memory" + else FileDesignationRepository(path) + ) + assert repo.entries() == rows + repo.entries().clear() + assert repo.entries() == rows + + +def test_missing_or_invalid_designations_do_not_invent_exposure(tmp_path): + path = tmp_path / "designations.json" + repo = FileDesignationRepository(path) + assert repo.entries() == [] + path.write_text("{invalid") + assert repo.entries() == [] + path.write_text('{"designations": [{"match": "x", "exposure": "not-a-designation"}]}') + assert repo.entries() == [] diff --git a/tests/v1/services/test_portfolio_observation.py b/tests/v1/services/test_portfolio_observation.py new file mode 100644 index 00000000..22245303 --- /dev/null +++ b/tests/v1/services/test_portfolio_observation.py @@ -0,0 +1,125 @@ +from datetime import timedelta + +import pytest + +from tests.v1.observation_support import NOW, REFERENCE, observer, repository_id_for +from traust_core.v1.errors import NotFoundError +from traust_core.v1.models.observation import ComparisonStatus +from traust_core.v1.providers.clock import FixedClock +from traust_core.v1.services.portfolio import BanBreaker + +RID = repository_id_for(None, "test") + + +@pytest.mark.parametrize("case", REFERENCE["cases"], ids=lambda c: c["name"]) +def test_recorded_legacy_observation_parity(case): + service, _, _, _, source = observer(case) + service.record_forge_facts(repo_url=case["repo"]["repo_url"]) + state = service.observed(repo_url=case["repo"]["repo_url"]) + assert state.status == case["status"] + for new, old in { + "changed_lines": "C", + "sensitive_changed": "S", + "sensitive_lines": "S_lines", + "deps_only": "deps_only", + "ahead_by": "ahead_by", + "compare_truncated": "truncated", + }.items(): + assert getattr(state, new) == case["observed"][old] + for field, old in { + "risk_tier": "tier", + "exposure": "exposure", + "audit_age_days": "audit_age_days", + }.items(): + assert getattr(state, field) == case["ctx"][old] + assert bool(state.push_changed) == case["ctx"]["changed"] + assert state.churn_ratio == (case["ctx"]["R"] if state.changed_lines is not None else None) + assert source.calls.count("compare") == int(case["info"]["ok"] and case["ctx"]["changed"]) + + +def test_state_recomputes_findings_and_age_without_contacting_forge(): + service, store, _assessment, findings, source = observer(live_crit_high=5) + service.record_forge_facts(repository_id=RID) + before = service.observed(repository_id=RID) + stored = dict(store.rows) + calls = list(source.calls) + findings.counts.clear() + service._clock = service.query._clock = FixedClock(NOW + timedelta(days=3)) + after = service.observed(repository_id=RID) + assert before.risk_tier == "P0" and after.risk_tier == "P2" + assert after.audit_age_days == before.audit_age_days + 3 + assert store.rows == stored + assert source.calls == calls + + +@pytest.mark.parametrize( + "field,value", [("pinned_sha", "b" * 40), ("audit_date", "2026-10-02"), ("has_audit", False)] +) +def test_changed_baseline_cannot_reuse_comparison(field, value): + service, _, assessment, _, _ = observer() + service.record_forge_facts(repository_id=RID) + assessment.target = assessment.target.model_copy(update={field: value}) + with pytest.raises(NotFoundError, match="baseline changed"): + service.observed(repository_id=RID) + + +def test_new_lines_reviewed_recomputes_churn_without_refetch(): + service, _, assessment, _, source = observer(lines_reviewed=100) + service.record_forge_facts(repository_id=RID) + before = service.observed(repository_id=RID) + assessment.target = assessment.target.model_copy(update={"lines_reviewed": 200}) + assert service.observed(repository_id=RID).churn_ratio == before.churn_ratio / 2 + assert source.calls.count("compare") == 1 + + +@pytest.mark.parametrize("status", list(ComparisonStatus)) +def test_status_is_stored_and_reassembled(status): + changes, kwargs = {}, {} + if status == ComparisonStatus.NO_REPO_URL: + changes["repo_url"] = None + elif status == ComparisonStatus.UNSUPPORTED_HOST: + changes["repo_url"] = "https://unsupported.example/o/r" + elif status == ComparisonStatus.NEVER_AUDITED: + changes["has_audit"] = False + elif status == ComparisonStatus.NO_PINNED_SHA: + changes["pinned_sha"] = None + elif status == ComparisonStatus.NETWORK_SKIPPED: + kwargs["network"] = False + elif status == ComparisonStatus.QUOTA_DEFERRED: + kwargs["quota_deferred"] = True + elif status == ComparisonStatus.BAN_SUSPECTED: + breaker = BanBreaker(1) + breaker.record(False) + kwargs["breaker"] = breaker + service, _, _, _, source = observer(**changes) + if status in ( + ComparisonStatus.NO_CREDENTIALS, + ComparisonStatus.UNREACHABLE, + ComparisonStatus.ERROR, + ): + source.info_data = {"ok": False, "kind": status} + facts = service.record_forge_facts(repository_id=RID, **kwargs) + assert facts.status == service.observed(repository_id=RID).status == status + + +def test_no_network_never_contacts_source(): + service, _, _, _, source = observer() + service.record_forge_facts(repository_id=RID, network=False) + assert source.calls == [] + + +def test_batch_quota_decision_does_not_query_live_quota(): + service, _, _, _, source = observer() + service.record_forge_facts(repository_id=RID, quota_deferred=False) + assert source.calls == ["info", "compare"] + + +def test_metric_programming_errors_are_not_degraded(monkeypatch): + service, _, _, _, _ = observer() + + def broken(files): + raise TypeError("metric bug") + + monkeypatch.setattr("traust_core.v1.services.observation.change_metrics", broken) + with pytest.raises(TypeError, match="metric bug"): + service.record_forge_facts(repository_id=RID) From 5742d55b6f221ace1a222c47a31ed1c883448a33 Mon Sep 17 00:00:00 2001 From: Shaun Smith <51304449+smith-xyz@users.noreply.github.com> Date: Wed, 21 Oct 2026 13:04:58 -0400 Subject: [PATCH 2/2] fix(context, repositories): the three reproduced defects from re-review - repository_observation is a Context method again (the splice had moved it onto the _Observation dataclass), and the door now opens one session unit of work so collaborators share a live connection - SqlUnitOfWork is re-entrant: nested with-blocks share the outermost connection and transaction; the outer owner rolls back and closes - observation_target reads only completed assessments as baselines - a degraded or failed run is never read as clean (the DDL rule) - via a status filter on AssessmentRepository.latest Regression tests: the door end-to-end over the contract tables, nested unit-of-work sharing, and failed-audit baseline selection. --- src/traust_core/v1/context/context.py | 73 ++++++++++--------- src/traust_core/v1/models/assessment.py | 4 + src/traust_core/v1/repositories/sql.py | 20 +++-- .../traust/assessments/assessment.py | 13 +++- src/traust_core/v1/services/assessment.py | 6 +- tests/v1/context/test_observation.py | 38 ++++++++++ .../tables/test_observation_target.py | 67 ++++++++++++----- tests/v1/repositories/test_sql.py | 17 +++++ 8 files changed, 174 insertions(+), 64 deletions(-) diff --git a/src/traust_core/v1/context/context.py b/src/traust_core/v1/context/context.py index 6f17f179..eda850cc 100644 --- a/src/traust_core/v1/context/context.py +++ b/src/traust_core/v1/context/context.py @@ -51,6 +51,15 @@ class Services: ... +@dataclass(frozen=True) +class _Observation: + """What the observation door yields: the write path, the composite read, the store.""" + + service: PortfolioService + query: RepoStateQuery + store: InMemoryObservationStore + + class Context: def __init__( self, @@ -118,15 +127,6 @@ def services(self) -> Services: def analysis_results(self) -> AnalysisResultsRepository: return AnalysisResultsRepository(self.object_store, ARTIFACT_FOR_KIND) - -@dataclass(frozen=True) -class _Observation: - """What the observation door yields: the write path, the composite read, the store.""" - - service: PortfolioService - query: RepoStateQuery - store: InMemoryObservationStore - @contextmanager def repository_observation( self, @@ -135,33 +135,34 @@ def repository_observation( ) -> Iterator[_Observation]: """Observe repositories: forge facts in, derived routing state out. - Reads inventory, assessments and open counts from the contract tables; - the observation store is in memory until contracts lands - repository_observation (NEED marker in repositories/observation.py). + Reads inventory, assessments and open counts from the contract tables + over one session unit of work; the observation store is in memory until + contracts lands repository_observation (NEED marker in + repositories/observation.py). """ - uow = SqlTraustUnitOfWork(self.database) store = InMemoryObservationStore() policy = PortfolioPolicy() - assessment = AssessmentService(uow, clock=self.clock, hosts=self.config.forge_hosts) - service = PortfolioService( - uow, - policy, - store=store, - assessment=assessment, - sources=command_source(self.config.forge_hosts, clock=self.clock), - clock=self.clock, - hosts=self.config.forge_hosts, - ) - query = RepoStateQuery( - service, - assessment, - None, # findings service: open counts come from the finding_status view - uow.open_counts, - policy, - designations=( - FileDesignationRepository(designations) if designations is not None else None - ), - clock=self.clock, - hosts=self.config.forge_hosts, - ) - yield _Observation(service=service, query=query, store=store) + with SqlTraustUnitOfWork(self.database) as uow: + assessment = AssessmentService(uow, clock=self.clock, hosts=self.config.forge_hosts) + service = PortfolioService( + uow, + policy, + store=store, + assessment=assessment, + sources=command_source(self.config.forge_hosts, clock=self.clock), + clock=self.clock, + hosts=self.config.forge_hosts, + ) + query = RepoStateQuery( + service, + assessment, + None, # findings service: open counts come from the finding_status view + uow.open_counts, + policy, + designations=( + FileDesignationRepository(designations) if designations is not None else None + ), + clock=self.clock, + hosts=self.config.forge_hosts, + ) + yield _Observation(service=service, query=query, store=store) diff --git a/src/traust_core/v1/models/assessment.py b/src/traust_core/v1/models/assessment.py index b0b564cb..481829ea 100644 --- a/src/traust_core/v1/models/assessment.py +++ b/src/traust_core/v1/models/assessment.py @@ -74,6 +74,10 @@ def ratio(self) -> float: raise NotImplementedError("stub: replaces build_coverage_gap_rollup.build_rollup") +# NEED(contracts): the assessment-status enum, once it lands, owns these values. +ASSESSMENT_COMPLETED = "completed" + + class Assessment(AppendOnly): """3a · what was examined: a ref (and image), by an assessor, with what outcome.""" diff --git a/src/traust_core/v1/repositories/sql.py b/src/traust_core/v1/repositories/sql.py index e88508a5..a36015ac 100644 --- a/src/traust_core/v1/repositories/sql.py +++ b/src/traust_core/v1/repositories/sql.py @@ -152,9 +152,14 @@ def apply_schema(engine: Engine, statements: Iterable[str]) -> None: class SqlUnitOfWork: + """Re-entrant: collaborating services share one instance; the outermost `with` + owns the connection and the rollback. + """ + def __init__(self, engine: Engine) -> None: self._engine = engine self._connection: Connection | None = None + self._depth = 0 @property def connection(self) -> Connection: @@ -166,11 +171,13 @@ def _open_repositories(self) -> None: pass def __enter__(self) -> Self: - try: - self._connection = self._engine.connect() - except SQLAlchemyError as e: - raise RepositoryError(f"{type(self).__name__}: connect failed: {e}") from e - self._open_repositories() + if self._depth == 0: + try: + self._connection = self._engine.connect() + except SQLAlchemyError as e: + raise RepositoryError(f"{type(self).__name__}: connect failed: {e}") from e + self._open_repositories() + self._depth += 1 return self def __exit__( @@ -179,6 +186,9 @@ def __exit__( exc: BaseException | None, tb: TracebackType | None, ) -> None: + self._depth -= 1 + if self._depth > 0: + return # nested use: the outer owner owns the transaction try: self.rollback() finally: diff --git a/src/traust_core/v1/repositories/traust/assessments/assessment.py b/src/traust_core/v1/repositories/traust/assessments/assessment.py index f08dcc40..1d5fd5bb 100644 --- a/src/traust_core/v1/repositories/traust/assessments/assessment.py +++ b/src/traust_core/v1/repositories/traust/assessments/assessment.py @@ -33,7 +33,9 @@ class AssessmentRepository(RecordRepository[Assessment], Protocol): - def latest(self, repository_ref_id: UUID, kind: str) -> Assessment | None: ... + def latest( + self, repository_ref_id: UUID, kind: str, status: str | None = None + ) -> Assessment | None: ... def list_for_ref(self, repository_ref_id: UUID) -> list[Assessment]: ... def list_for_run_order(self, run_order_id: UUID) -> list[Assessment]: ... @@ -42,11 +44,16 @@ class SqlAssessmentRepository(SqlAppendOnlyRepository[Assessment]): spec = ASSESSMENT record = Assessment - def latest(self, repository_ref_id: UUID, kind: str) -> Assessment | None: + def latest( + self, repository_ref_id: UUID, kind: str, status: str | None = None + ) -> Assessment | None: c = self.table.c + conditions = [c.repository_ref_id == repository_ref_id, c.kind == kind] + if status is not None: + conditions.append(c.status == status) stmt = ( self._select() - .where(c.repository_ref_id == repository_ref_id, c.kind == kind) + .where(*conditions) .order_by(c.occurred_at.desc(), c.created_at.desc()) .limit(1) ) diff --git a/src/traust_core/v1/services/assessment.py b/src/traust_core/v1/services/assessment.py index 52f7cdd3..fa991525 100644 --- a/src/traust_core/v1/services/assessment.py +++ b/src/traust_core/v1/services/assessment.py @@ -7,6 +7,7 @@ from traust_core.v1.errors import NotFoundError, ValidationError from traust_core.v1.models.artifacts import AssessmentArtifact from traust_core.v1.models.assessment import ( + ASSESSMENT_COMPLETED, Assessment, AssessmentFilter, AssessmentKind, @@ -113,7 +114,10 @@ def observation_target( raise NotFoundError(f"repository not in inventory: {repository_id or repo_url}") latest = None for ref in uow.repository_refs.list_for(repo.id): - found = uow.assessments.latest(ref.id, AssessmentKind.SECURE_CODE) + # a degraded or failed run is never read as clean (the DDL rule) + found = uow.assessments.latest( + ref.id, AssessmentKind.SECURE_CODE, status=ASSESSMENT_COMPLETED + ) if found is not None and ( latest is None or found.occurred_at > latest[1].occurred_at ): diff --git a/tests/v1/context/test_observation.py b/tests/v1/context/test_observation.py index bb5c430a..9e054535 100644 --- a/tests/v1/context/test_observation.py +++ b/tests/v1/context/test_observation.py @@ -22,3 +22,41 @@ def run(self, argv, **kwargs): assert calls == [] assert source.info(HttpsRepoUrl("github.com", "o/r")).acquisition.acquired_at == NOW assert len(calls) == 1 + + +def test_observation_door_reads_the_contract_tables(tmp_path): + from datetime import UTC, datetime + + from tests.v1.conftest import contract_db + from tests.v1.repositories.tables.conftest import Seed + from traust_core.v1.context.config import Config, DatabaseConfig + from traust_core.v1.context.context import Context + from traust_core.v1.models.assessment import AssessmentKind + from traust_core.v1.providers.clock import FixedClock + from traust_core.v1.repositories.contract import SqlTraustUnitOfWork + + with contract_db("sqlite", tmp_path) as engine: + pass # schema applied to a fresh sqlite file; reuse the same path + url = str(engine.url) + engine.dispose() + with SqlTraustUnitOfWork(engine) as uow: + seed = Seed(uow) + repo = seed.repository(url="https://github.com/org/door") + ref = seed.repository_ref(repository=repo, sha="c" * 40) + seed.assessment( + ref=ref, + kind=AssessmentKind.SECURE_CODE, + occurred_at=datetime(2026, 9, 1, tzinfo=UTC), + ) + uow.commit() + engine.dispose() + + config = Config(database=DatabaseConfig(url=url)) + context = Context(config, clock=FixedClock(NOW)) + + with context.repository_observation() as obs: + facts = obs.service.record_forge_facts(repository_id=repo.id, network=False) + assert facts.repository_id == repo.id + state = obs.query.observed(repository_id=repo.id) + assert state.repository_id == repo.id + assert obs.store.find(repo.id).repository_id == repo.id diff --git a/tests/v1/repositories/tables/test_observation_target.py b/tests/v1/repositories/tables/test_observation_target.py index ba493a3a..798efb28 100644 --- a/tests/v1/repositories/tables/test_observation_target.py +++ b/tests/v1/repositories/tables/test_observation_target.py @@ -6,7 +6,6 @@ from uuid import UUID import pytest -from sqlalchemy.engine import Engine from tests.v1.repositories.tables.conftest import Seed from traust_core.v1.errors import NotFoundError, ValidationError @@ -16,12 +15,12 @@ from traust_core.v1.services.assessment import AssessmentService -def service(engine: Engine) -> AssessmentService: - """A service over its own unit of work, as the context door builds it.""" - return AssessmentService(SqlTraustUnitOfWork(engine)) +def service(uow: SqlTraustUnitOfWork) -> AssessmentService: + """A service sharing the open unit of work, as the observation door wires it.""" + return AssessmentService(uow) -def test_resolves_latest_audit_and_pinned_commit(uow, seed: Seed, engine: Engine): +def test_resolves_latest_audit_and_pinned_commit(uow, seed: Seed): repo = seed.repository(url="https://github.com/org/audited") old_ref = seed.repository_ref(repository=repo, sha="a" * 40) new_ref = seed.repository_ref(repository=repo, sha="b" * 40) @@ -41,9 +40,7 @@ def test_resolves_latest_audit_and_pinned_commit(uow, seed: Seed, engine: Engine kind=AssessmentKind.THREAT_MODEL, occurred_at=datetime(2026, 10, 1, tzinfo=UTC), ) - uow.commit() - - target = service(engine).observation_target(repo_url="https://github.com/org/audited") + target = service(uow).observation_target(repo_url="https://github.com/org/audited") assert target == ObservationTarget( repository_id=repo.id, repo_url="https://github.com/org/audited", @@ -52,34 +49,66 @@ def test_resolves_latest_audit_and_pinned_commit(uow, seed: Seed, engine: Engine pinned_sha="b" * 40, repository_ref_id=new_ref.id, ) - by_id = service(engine).observation_target(repository_id=repo.id) + by_id = service(uow).observation_target(repository_id=repo.id) assert by_id == target -def test_observation_target_without_an_audit_is_never_audited(uow, seed: Seed, engine: Engine): - repo = seed.repository(url="https://github.com/org/never-audited") - uow.commit() +def test_a_failed_audit_is_never_the_baseline(uow, seed: Seed): + repo = seed.repository(url="https://github.com/org/flaky") + ref = seed.repository_ref(repository=repo, sha="a" * 40) + seed.assessment( + ref=ref, + kind=AssessmentKind.SECURE_CODE, + occurred_at=datetime(2026, 9, 1, tzinfo=UTC), + ) + # a newer run that failed must not displace the completed baseline + seed.assessment( + ref=ref, + kind=AssessmentKind.SECURE_CODE, + status="failed", + occurred_at=datetime(2026, 10, 1, tzinfo=UTC), + ) + + target = service(uow).observation_target(repo_url="https://github.com/org/flaky") + assert target.audit_date == "2026-09-01" - target = service(engine).observation_target(repo_url="https://github.com/org/never-audited") + +def test_only_failed_audits_count_as_never_audited(uow, seed: Seed): + repo = seed.repository(url="https://github.com/org/all-failed") + ref = seed.repository_ref(repository=repo, sha="a" * 40) + seed.assessment( + ref=ref, + kind=AssessmentKind.SECURE_CODE, + status="failed", + occurred_at=datetime(2026, 10, 1, tzinfo=UTC), + ) + + target = service(uow).observation_target(repo_url="https://github.com/org/all-failed") + assert target.has_audit is False + + +def test_observation_target_without_an_audit_is_never_audited(uow, seed: Seed): + repo = seed.repository(url="https://github.com/org/never-audited") + target = service(uow).observation_target(repo_url="https://github.com/org/never-audited") assert target.has_audit is False assert target.repository_id == repo.id -def test_observation_target_requires_inventory_membership(uow, seed: Seed, engine: Engine): +def test_observation_target_requires_inventory_membership(uow, seed: Seed): with pytest.raises(NotFoundError): - service(engine).observation_target(repo_url="https://github.com/org/unknown") + service(uow).observation_target(repo_url="https://github.com/org/unknown") -def test_observation_target_takes_exactly_one_selector(engine: Engine): +def test_observation_target_takes_exactly_one_selector(uow): with pytest.raises(ValidationError): - service(engine).observation_target() + service(uow).observation_target() with pytest.raises(ValidationError): - service(engine).observation_target( + service(uow).observation_target( repository_id=UUID(int=1), repo_url="https://github.com/org/x" ) -def test_open_counts_read_the_finding_status_view(uow, seed: Seed, engine: Engine): +def test_open_counts_read_the_finding_status_view(uow, seed: Seed): from traust_core.v1.models.vocabulary import Severity repo = seed.repository() diff --git a/tests/v1/repositories/test_sql.py b/tests/v1/repositories/test_sql.py index 7ac5fbf8..52e3ad2f 100644 --- a/tests/v1/repositories/test_sql.py +++ b/tests/v1/repositories/test_sql.py @@ -266,3 +266,20 @@ def test_saved_record_round_trips_by_id(uow: RecordUnitOfWork) -> None: with uow as u: reread = u.records.get(saved.id) assert reread == saved + + +def test_unit_of_work_is_reentrant(engine) -> None: + from traust_core.v1.repositories.sql import SqlUnitOfWork + + outer = SqlUnitOfWork(engine) + with outer as first: + with outer as second: # a collaborating service re-enters the same unit + assert second is first + assert second.connection is first.connection + # the nested exit must not close the session connection + assert first.connection.execute(sa.text("SELECT 1")).scalar() == 1 + # only the outermost exit closes + from traust_core.v1.errors import RepositoryError + + with pytest.raises(RepositoryError): + _ = first.connection