1919 lines
70 KiB
Python
1919 lines
70 KiB
Python
"""Bounded installed-composition and externally authorized fit evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
import hashlib
|
|
import hmac
|
|
from importlib import metadata
|
|
from importlib.util import source_from_cache
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import stat
|
|
import sys
|
|
from typing import Any, Iterable, Mapping, Sequence
|
|
from urllib.parse import urlsplit
|
|
|
|
from cryptography.exceptions import InvalidSignature
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
from jsonschema.exceptions import SchemaError
|
|
|
|
|
|
MAX_GOVOPLAN_DISTRIBUTIONS = 256
|
|
MAX_DISTRIBUTION_FILES = 10_000
|
|
MAX_COLLECTION_RECORD_FILES = 50_000
|
|
MAX_HASHED_FILE_BYTES = 64 * 1024 * 1024
|
|
MAX_HASHED_DISTRIBUTION_BYTES = 512 * 1024 * 1024
|
|
MAX_HASHED_COLLECTION_BYTES = 2 * 1024 * 1024 * 1024
|
|
MAX_DIRECT_URL_BYTES = 64 * 1024
|
|
MAX_LOCAL_OBSERVATION_AGE = timedelta(minutes=5)
|
|
MAX_LOCAL_OBSERVATION_FUTURE_SKEW = timedelta(seconds=30)
|
|
INSTALLED_OBSERVATION_MODES = frozenset({"direct_local", "imported_unsigned"})
|
|
VERIFICATION_MODES = frozenset({"current", "historical_override"})
|
|
PACKAGE_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
OPAQUE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$")
|
|
VERSION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+!-]*$")
|
|
ENTRY_POINT_VALUE_PATTERN = re.compile(
|
|
r"^[A-Za-z_][A-Za-z0-9_.]*(?::[A-Za-z_][A-Za-z0-9_.]*)?(?:\s*\[[A-Za-z0-9_., -]+\])?$"
|
|
)
|
|
VCS_REVISION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/+~-]{0,159}$")
|
|
GIT_OBJECT_PATTERN = re.compile(r"^[0-9a-f]{40,64}$")
|
|
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
|
|
POSITIVE_RESULTS = {
|
|
"target_environment": "passed",
|
|
"external_providers": "passed",
|
|
"production_approval": "approved",
|
|
}
|
|
NEGATIVE_RESULTS = {
|
|
"target_environment": "failed",
|
|
"external_providers": "failed",
|
|
"production_approval": "rejected",
|
|
}
|
|
PROOF_REQUIREMENTS = {
|
|
"target_environment": (
|
|
"A target-run result bound to this assessment release and installed-evidence digest.",
|
|
"An independently trusted authority key permitted for target_environment proof.",
|
|
"Opaque control IDs and content hashes for the target acceptance artifacts.",
|
|
),
|
|
"external_providers": (
|
|
"Provider acceptance results bound to this assessment release and installed-evidence digest.",
|
|
"An independently trusted authority key permitted for external_providers proof.",
|
|
"Opaque control IDs and content hashes; credentials and endpoint identifiers stay outside the report.",
|
|
),
|
|
"production_approval": (
|
|
"An explicit approval bound to this assessment release and installed-evidence digest.",
|
|
"An independently provisioned authority key permitted for production_approval.",
|
|
"An unexpired signed proof bundle; an assessment or operator cannot approve itself.",
|
|
),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class EvidenceFinding:
|
|
severity: str
|
|
code: str
|
|
message: str
|
|
assessment_ids: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class InstalledEvidenceReview:
|
|
findings: tuple[EvidenceFinding, ...]
|
|
changes: tuple[dict[str, Any], ...]
|
|
changed_repositories: frozenset[str]
|
|
affected_module_ids: frozenset[str]
|
|
proof_scope: Mapping[str, Any]
|
|
evidence_sha256: str | None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class BoundaryEvidenceReview:
|
|
findings: tuple[EvidenceFinding, ...]
|
|
proof_scope: Mapping[str, Any]
|
|
review_targets: tuple[dict[str, Any], ...] = ()
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CollectionHashBudget:
|
|
remaining_files: int = MAX_COLLECTION_RECORD_FILES
|
|
remaining_bytes: int = MAX_HASHED_COLLECTION_BYTES
|
|
|
|
|
|
def validate_payload(
|
|
*, payload: dict[str, Any], schema: dict[str, Any]
|
|
) -> tuple[str, ...]:
|
|
try:
|
|
Draft202012Validator.check_schema(schema)
|
|
except SchemaError as exc:
|
|
return (f"$schema: {exc.message}",)
|
|
validator = Draft202012Validator(schema, format_checker=FormatChecker())
|
|
return tuple(
|
|
f"{_json_path(error.absolute_path)}: failed {error.validator!r} validation"
|
|
for error in sorted(
|
|
validator.iter_errors(payload),
|
|
key=lambda item: (
|
|
tuple(str(part) for part in item.absolute_path),
|
|
str(item.validator),
|
|
item.message,
|
|
),
|
|
)
|
|
)
|
|
|
|
|
|
def collect_installed_composition(
|
|
*,
|
|
assessment: dict[str, Any],
|
|
collected_at: datetime | None = None,
|
|
distributions: Iterable[metadata.Distribution] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Collect only bounded, non-identifying GovOPlaN distribution evidence.
|
|
|
|
Loading ``govoplan.modules`` entry points executes installed module code. The
|
|
collector catches failures and records only a stable error code; it never
|
|
serializes exception text, URLs, local paths, hostnames, or usernames.
|
|
"""
|
|
|
|
artifacts: list[dict[str, Any]] = []
|
|
issues: list[dict[str, str]] = []
|
|
hash_budget = CollectionHashBudget()
|
|
candidates: list[tuple[str, metadata.Distribution]] = []
|
|
source = metadata.distributions() if distributions is None else distributions
|
|
for distribution in source:
|
|
try:
|
|
raw_package_name = distribution.metadata["Name"]
|
|
if not isinstance(raw_package_name, str) or len(raw_package_name) > 256:
|
|
continue
|
|
package_name = normalize_package_name(raw_package_name)
|
|
except (Exception, SystemExit):
|
|
continue
|
|
if (
|
|
package_name.startswith("govoplan-")
|
|
and len(package_name) <= 128
|
|
and PACKAGE_PATTERN.fullmatch(package_name) is not None
|
|
):
|
|
candidates.append((package_name, distribution))
|
|
|
|
candidates.sort(key=lambda item: (item[0], _distribution_version(item[1])))
|
|
if len(candidates) > MAX_GOVOPLAN_DISTRIBUTIONS:
|
|
candidates = candidates[:MAX_GOVOPLAN_DISTRIBUTIONS]
|
|
issues.append(
|
|
{
|
|
"code": "collection-limit-exceeded",
|
|
"package_name": "govoplan-collector",
|
|
}
|
|
)
|
|
|
|
name_counts: dict[str, int] = {}
|
|
for package_name, _ in candidates:
|
|
name_counts[package_name] = name_counts.get(package_name, 0) + 1
|
|
for package_name, count in sorted(name_counts.items()):
|
|
if count > 1:
|
|
issues.append(
|
|
{
|
|
"code": "duplicate-distribution",
|
|
"package_name": package_name,
|
|
}
|
|
)
|
|
|
|
for package_name, distribution in candidates:
|
|
version = _distribution_version(distribution)
|
|
if not version:
|
|
issues.append(
|
|
{
|
|
"code": "distribution-metadata-invalid",
|
|
"package_name": package_name
|
|
if PACKAGE_PATTERN.fullmatch(package_name)
|
|
else "govoplan-invalid",
|
|
}
|
|
)
|
|
continue
|
|
modules, module_issues = _distribution_modules(
|
|
package_name=package_name,
|
|
distribution=distribution,
|
|
)
|
|
issues.extend(module_issues)
|
|
artifacts.append(
|
|
{
|
|
"package_name": package_name,
|
|
"package_version": version,
|
|
"modules": modules,
|
|
"source_provenance": _source_provenance(distribution),
|
|
"record_integrity": _record_integrity(
|
|
distribution,
|
|
package_name=package_name,
|
|
budget=hash_budget,
|
|
),
|
|
}
|
|
)
|
|
|
|
observed_at = collected_at or datetime.now(tz=UTC)
|
|
if observed_at.tzinfo is None:
|
|
observed_at = observed_at.replace(tzinfo=UTC)
|
|
release = assessment.get("release")
|
|
assessment_release = release.get("ref") if isinstance(release, dict) else None
|
|
sorted_issues = sorted(
|
|
_unique_dicts(issues),
|
|
key=lambda item: (item["package_name"], item["code"]),
|
|
)
|
|
if len(sorted_issues) > 256:
|
|
limit_issue = {
|
|
"code": "collection-limit-exceeded",
|
|
"package_name": "govoplan-collector",
|
|
}
|
|
sorted_issues = sorted_issues[:255]
|
|
if limit_issue not in sorted_issues:
|
|
sorted_issues.append(limit_issue)
|
|
sorted_issues.sort(key=lambda item: (item["package_name"], item["code"]))
|
|
return {
|
|
"$schema": "./installed-composition-evidence.schema.json",
|
|
"schema_version": "0.2.0",
|
|
"evidence_kind": "govoplan.installed-composition",
|
|
"assessment_id": _safe_opaque_id(
|
|
assessment.get("assessment_id"), fallback="invalid-assessment"
|
|
),
|
|
"assessment_release": _safe_opaque_id(
|
|
assessment_release, fallback="invalid-release"
|
|
),
|
|
"collected_at": observed_at.astimezone(UTC).isoformat().replace("+00:00", "Z"),
|
|
"scope": "current-python-environment.govoplan-distributions",
|
|
"artifacts": sorted(
|
|
artifacts,
|
|
key=lambda item: (
|
|
item["package_name"],
|
|
item["package_version"],
|
|
canonical_bytes(item),
|
|
),
|
|
),
|
|
"collection_issues": sorted_issues,
|
|
}
|
|
|
|
|
|
def review_installed_composition(
|
|
*,
|
|
assessment: dict[str, Any],
|
|
catalog_entries: Mapping[str, Mapping[str, Any]],
|
|
selected_versions: Mapping[str, str],
|
|
selected_commits: Mapping[str, str],
|
|
evidence: dict[str, Any] | None,
|
|
schema: dict[str, Any] | None,
|
|
observation_mode: str = "imported_unsigned",
|
|
verification_time: datetime | None = None,
|
|
verification_mode: str = "current",
|
|
) -> InstalledEvidenceReview:
|
|
if evidence is None:
|
|
return InstalledEvidenceReview(
|
|
findings=(),
|
|
changes=(),
|
|
changed_repositories=frozenset(),
|
|
affected_module_ids=frozenset(),
|
|
proof_scope=_unchecked_installed_scope(),
|
|
evidence_sha256=None,
|
|
)
|
|
digest = canonical_sha256(evidence)
|
|
if schema is None:
|
|
finding = EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_schema_missing",
|
|
"Installed evidence was supplied without its validation schema.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
return InstalledEvidenceReview(
|
|
findings=(finding,),
|
|
changes=(),
|
|
changed_repositories=frozenset(),
|
|
affected_module_ids=frozenset(),
|
|
proof_scope=_invalid_installed_scope(digest=digest),
|
|
evidence_sha256=digest,
|
|
)
|
|
schema_errors = validate_payload(payload=evidence, schema=schema)
|
|
if schema_errors:
|
|
findings = tuple(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_schema",
|
|
message,
|
|
("assessment.installed_composition",),
|
|
)
|
|
for message in schema_errors
|
|
)
|
|
return InstalledEvidenceReview(
|
|
findings=findings,
|
|
changes=(),
|
|
changed_repositories=frozenset(),
|
|
affected_module_ids=frozenset(),
|
|
proof_scope=_invalid_installed_scope(digest=digest),
|
|
evidence_sha256=digest,
|
|
)
|
|
|
|
findings: list[EvidenceFinding] = []
|
|
effective_observation_mode = (
|
|
observation_mode
|
|
if observation_mode in INSTALLED_OBSERVATION_MODES
|
|
else "invalid"
|
|
)
|
|
effective_verification_mode = (
|
|
verification_mode if verification_mode in VERIFICATION_MODES else "invalid"
|
|
)
|
|
if effective_observation_mode == "invalid":
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_observation_mode",
|
|
"Installed evidence observation mode is invalid.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
if effective_verification_mode == "invalid":
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_verification_mode",
|
|
"Installed evidence verification mode is invalid.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
changes: list[dict[str, Any]] = []
|
|
changed_repositories: set[str] = set()
|
|
affected_module_ids: set[str] = set()
|
|
now = verification_time or datetime.now(tz=UTC)
|
|
if now.tzinfo is None:
|
|
now = now.replace(tzinfo=UTC)
|
|
collected_at = _datetime(evidence.get("collected_at"))
|
|
observation_trusted = False
|
|
freshness = "invalid"
|
|
if effective_observation_mode == "direct_local":
|
|
if collected_at is None:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_time_invalid",
|
|
"Direct installed evidence has no valid collection time.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
elif collected_at > now + MAX_LOCAL_OBSERVATION_FUTURE_SKEW:
|
|
freshness = "future"
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_from_future",
|
|
"Direct installed evidence is newer than the permitted clock skew.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
elif now - collected_at > MAX_LOCAL_OBSERVATION_AGE:
|
|
freshness = "stale"
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_stale",
|
|
"Direct installed evidence is older than the five-minute acceptance window.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
else:
|
|
freshness = "fresh"
|
|
observation_trusted = True
|
|
elif effective_observation_mode == "imported_unsigned":
|
|
freshness = "unsigned_import"
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_unsigned_import",
|
|
"Imported unsigned installed evidence may be compared but cannot establish installed proof.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
if effective_verification_mode == "invalid":
|
|
observation_trusted = False
|
|
expected_release = (
|
|
assessment.get("release", {}).get("ref")
|
|
if isinstance(assessment.get("release"), dict)
|
|
else None
|
|
)
|
|
binding_valid = True
|
|
if evidence.get("assessment_id") != assessment.get("assessment_id"):
|
|
binding_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_assessment_mismatch",
|
|
"Installed evidence is bound to a different assessment ID.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
if evidence.get("assessment_release") != expected_release:
|
|
binding_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_evidence_release_mismatch",
|
|
"Installed evidence is bound to a different assessment release.",
|
|
("assessment.release", "assessment.installed_composition"),
|
|
)
|
|
)
|
|
|
|
artifact_rows = [
|
|
item for item in evidence.get("artifacts", []) if isinstance(item, dict)
|
|
]
|
|
artifacts_by_package: dict[str, dict[str, Any]] = {}
|
|
duplicate_packages: set[str] = set()
|
|
for artifact in artifact_rows:
|
|
package_name = normalize_package_name(str(artifact.get("package_name") or ""))
|
|
if package_name in artifacts_by_package:
|
|
duplicate_packages.add(package_name)
|
|
else:
|
|
artifacts_by_package[package_name] = artifact
|
|
for package_name in sorted(duplicate_packages):
|
|
binding_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_distribution_duplicate",
|
|
f"Installed evidence repeats distribution {package_name!r}.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
|
|
expected_by_package: dict[str, tuple[dict[str, Any], Mapping[str, Any]]] = {}
|
|
assessment_by_module: dict[str, dict[str, Any]] = {}
|
|
for component in assessment.get("composition", []):
|
|
if not isinstance(component, dict):
|
|
continue
|
|
module_id = str(component.get("module_id") or "")
|
|
assessment_by_module[module_id] = component
|
|
if component.get("enabled") is not True:
|
|
continue
|
|
catalog_entry = catalog_entries.get(module_id)
|
|
if catalog_entry is None:
|
|
binding_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_expected_catalog_entry_missing",
|
|
f"Enabled module {module_id!r} cannot be mapped to a catalog package.",
|
|
(f"composition.{module_id}",),
|
|
)
|
|
)
|
|
continue
|
|
package_name = normalize_package_name(str(catalog_entry.get("package") or ""))
|
|
if not package_name:
|
|
binding_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_expected_package_missing",
|
|
f"Enabled module {module_id!r} has no valid catalog package name.",
|
|
(f"composition.{module_id}",),
|
|
)
|
|
)
|
|
continue
|
|
if package_name in expected_by_package:
|
|
binding_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_expected_package_ambiguous",
|
|
f"Multiple enabled composition entries map to {package_name!r}.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
continue
|
|
expected_by_package[package_name] = (component, catalog_entry)
|
|
|
|
if not expected_by_package:
|
|
binding_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"installed_expected_composition_empty",
|
|
"Installed proof requires at least one enabled assessment component mapped to a catalog package.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
|
|
record_expected = 0
|
|
record_verified = 0
|
|
generated_unhashed_file_count = 0
|
|
provenance_expected = 0
|
|
provenance_anchored = 0
|
|
provenance_mutable = 0
|
|
provenance_unanchored = 0
|
|
artifact_valid = binding_valid and not duplicate_packages
|
|
|
|
for package_name, (component, catalog_entry) in sorted(expected_by_package.items()):
|
|
module_id = str(component["module_id"])
|
|
repository = str(component["repository"])
|
|
artifact = artifacts_by_package.get(package_name)
|
|
record_expected += 1
|
|
provenance_expected += 1
|
|
if package_name in duplicate_packages:
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_distribution_duplicate",
|
|
code="installed_distribution_duplicate_ambiguous",
|
|
message=f"Installed distribution {package_name!r} is duplicated, so no candidate was selected for integrity or provenance conclusions.",
|
|
)
|
|
continue
|
|
if artifact is None:
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_distribution_missing",
|
|
code="installed_distribution_missing",
|
|
message=f"Enabled module {module_id!r} has no installed distribution {package_name!r} in the evidence.",
|
|
)
|
|
continue
|
|
|
|
expected_version = str(component["manifest_version"])
|
|
package_version = str(artifact.get("package_version") or "")
|
|
catalog_version = str(catalog_entry.get("version") or "")
|
|
if package_version != expected_version or package_version != catalog_version:
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_version_mismatch",
|
|
code="installed_distribution_version_mismatch",
|
|
message=(
|
|
f"Installed {package_name!r} version {package_version!r} does not match "
|
|
f"assessment {expected_version!r} and catalog {catalog_version!r}."
|
|
),
|
|
extra={
|
|
"installed_version": package_version,
|
|
"assessed_version": expected_version,
|
|
"catalog_version": catalog_version,
|
|
},
|
|
)
|
|
|
|
modules = [
|
|
item for item in artifact.get("modules", []) if isinstance(item, dict)
|
|
]
|
|
module_rows: dict[str, dict[str, Any]] = {}
|
|
duplicate_module_ids: set[str] = set()
|
|
for item in modules:
|
|
observed_id = str(item.get("module_id") or "")
|
|
if observed_id in module_rows:
|
|
duplicate_module_ids.add(observed_id)
|
|
else:
|
|
module_rows[observed_id] = item
|
|
expected_manifest = module_rows.get(module_id)
|
|
manifest_mismatch = (
|
|
expected_manifest is None if module_id != "core" else False
|
|
) or (
|
|
expected_manifest is not None
|
|
and str(expected_manifest.get("manifest_version") or "") != expected_version
|
|
)
|
|
if manifest_mismatch:
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_manifest_mismatch",
|
|
code="installed_manifest_version_mismatch",
|
|
message=f"Installed distribution {package_name!r} does not expose module {module_id!r} at version {expected_version!r}.",
|
|
)
|
|
unexpected_modules = sorted(set(module_rows) - {module_id})
|
|
if duplicate_module_ids or unexpected_modules:
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_manifest_set_mismatch",
|
|
code="installed_manifest_set_mismatch",
|
|
message=f"Installed distribution {package_name!r} exposes an unexpected or duplicate module entry-point set.",
|
|
)
|
|
|
|
record = artifact.get("record_integrity")
|
|
record_status = record.get("status") if isinstance(record, dict) else None
|
|
record_semantics_valid = (
|
|
_record_integrity_semantics_valid(record)
|
|
if isinstance(record, dict)
|
|
else False
|
|
)
|
|
if isinstance(record, dict):
|
|
generated_count = record.get("generated_unhashed_file_count")
|
|
if isinstance(generated_count, int) and not isinstance(
|
|
generated_count, bool
|
|
):
|
|
generated_unhashed_file_count += max(generated_count, 0)
|
|
if record_status == "verified" and record_semantics_valid:
|
|
record_verified += 1
|
|
else:
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_record_unverified",
|
|
code=(
|
|
"installed_record_integrity_unverified"
|
|
if record_semantics_valid
|
|
else "installed_record_integrity_invalid"
|
|
),
|
|
message=(
|
|
f"Installed distribution {package_name!r} does not have complete, matching RECORD hash coverage."
|
|
if record_semantics_valid
|
|
else f"Installed distribution {package_name!r} has contradictory RECORD status and counters."
|
|
),
|
|
extra={"record_status": record_status},
|
|
)
|
|
|
|
provenance = artifact.get("source_provenance")
|
|
provenance_kind = (
|
|
str(provenance.get("kind") or "") if isinstance(provenance, dict) else ""
|
|
)
|
|
if provenance_kind == "vcs-commit":
|
|
commit = str(provenance.get("commit") or "").lower()
|
|
selected_version = selected_versions.get(repository)
|
|
selected_commit = selected_commits.get(repository)
|
|
assessed_commit = str(component.get("commit") or "").lower()
|
|
commit_matches = commit.startswith(assessed_commit)
|
|
selected_matches = (
|
|
selected_version != expected_version
|
|
or selected_commit is None
|
|
or commit == selected_commit.lower()
|
|
)
|
|
if commit_matches and selected_matches:
|
|
provenance_anchored += 1
|
|
else:
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_commit_mismatch",
|
|
code="installed_source_commit_mismatch",
|
|
message=f"Installed immutable VCS provenance for {package_name!r} does not match the assessed or signed selected-unit commit.",
|
|
)
|
|
elif provenance_kind in {"editable-vcs", "editable-local", "local-directory"}:
|
|
provenance_mutable += 1
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_source_mutable",
|
|
code="installed_source_mutable",
|
|
message=f"Installed distribution {package_name!r} is editable or directory-backed and is not immutable release evidence.",
|
|
)
|
|
else:
|
|
provenance_unanchored += 1
|
|
artifact_valid = False
|
|
_record_change(
|
|
findings=findings,
|
|
changes=changes,
|
|
changed_repositories=changed_repositories,
|
|
affected_module_ids=affected_module_ids,
|
|
component=component,
|
|
kind="installed_source_unanchored",
|
|
code="installed_source_unanchored",
|
|
message=f"Installed distribution {package_name!r} has no source provenance anchored to the assessed release.",
|
|
)
|
|
|
|
expected_packages = set(expected_by_package)
|
|
for package_name in sorted(set(artifacts_by_package) - expected_packages):
|
|
artifact_valid = False
|
|
matching_module_id = next(
|
|
(
|
|
module_id
|
|
for module_id, entry in catalog_entries.items()
|
|
if normalize_package_name(str(entry.get("package") or ""))
|
|
== package_name
|
|
),
|
|
None,
|
|
)
|
|
component = assessment_by_module.get(str(matching_module_id or ""))
|
|
repository = (
|
|
str(component.get("repository")) if isinstance(component, dict) else None
|
|
)
|
|
if repository:
|
|
changed_repositories.add(repository)
|
|
if matching_module_id and component is not None:
|
|
affected_module_ids.add(str(matching_module_id))
|
|
target_ids = (
|
|
(f"composition.{matching_module_id}",)
|
|
if matching_module_id and component is not None
|
|
else ("assessment.installed_composition",)
|
|
)
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"review",
|
|
"installed_distribution_extra",
|
|
f"Installed GovOPlaN distribution {package_name!r} is outside the enabled assessed composition.",
|
|
target_ids,
|
|
)
|
|
)
|
|
changes.append(
|
|
{
|
|
"module_id": str(matching_module_id or package_name),
|
|
"repository": repository,
|
|
"kind": "installed_distribution_extra",
|
|
"package_name": package_name,
|
|
}
|
|
)
|
|
|
|
for issue in evidence.get("collection_issues", []):
|
|
if not isinstance(issue, dict):
|
|
continue
|
|
package_name = str(issue.get("package_name") or "")
|
|
code = str(issue.get("code") or "")
|
|
artifact_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"review",
|
|
"installed_collection_incomplete",
|
|
f"Installed evidence collection reported {code!r} for {package_name!r}.",
|
|
("assessment.installed_composition",),
|
|
)
|
|
)
|
|
|
|
exact_set = (
|
|
set(artifacts_by_package) == expected_packages and not duplicate_packages
|
|
)
|
|
installed_checked = observation_trusted
|
|
record_checked = (
|
|
observation_trusted and binding_valid and exact_set and record_expected > 0
|
|
)
|
|
provenance_checked = (
|
|
observation_trusted and binding_valid and exact_set and provenance_expected > 0
|
|
)
|
|
proof_scope = {
|
|
"installed_artifacts": {
|
|
"checked": installed_checked,
|
|
"valid": artifact_valid if installed_checked else None,
|
|
"comparison_valid": artifact_valid,
|
|
"evidence_sha256": digest,
|
|
"expected_distribution_count": len(expected_packages),
|
|
"observed_distribution_count": len(artifact_rows),
|
|
},
|
|
"installed_record_integrity": {
|
|
"checked": record_checked,
|
|
"valid": record_verified == record_expected if record_checked else None,
|
|
"verified_distribution_count": record_verified,
|
|
"expected_distribution_count": record_expected,
|
|
"generated_unhashed_file_count": generated_unhashed_file_count,
|
|
},
|
|
"installed_source_provenance": {
|
|
"checked": provenance_checked,
|
|
"valid": provenance_anchored == provenance_expected
|
|
if provenance_checked
|
|
else None,
|
|
"anchored_distribution_count": provenance_anchored,
|
|
"mutable_distribution_count": provenance_mutable,
|
|
"unanchored_distribution_count": provenance_unanchored,
|
|
"expected_distribution_count": provenance_expected,
|
|
},
|
|
"runtime_activation": {
|
|
"checked": False,
|
|
"valid": None,
|
|
"required_evidence": (
|
|
"A separately collected runtime registry/configuration snapshot bound to the installed evidence digest.",
|
|
),
|
|
},
|
|
"installed_evidence_observation": {
|
|
"checked": True,
|
|
"mode": effective_observation_mode,
|
|
"freshness": freshness,
|
|
"collected_at": _iso_datetime(collected_at),
|
|
"evaluated_at": _iso_datetime(now),
|
|
"verification_mode": effective_verification_mode,
|
|
"freshness_window_seconds": int(MAX_LOCAL_OBSERVATION_AGE.total_seconds()),
|
|
},
|
|
}
|
|
return InstalledEvidenceReview(
|
|
findings=tuple(findings),
|
|
changes=tuple(changes),
|
|
changed_repositories=frozenset(changed_repositories),
|
|
affected_module_ids=frozenset(affected_module_ids),
|
|
proof_scope=proof_scope,
|
|
evidence_sha256=digest,
|
|
)
|
|
|
|
|
|
def review_boundary_evidence(
|
|
*,
|
|
assessment: dict[str, Any],
|
|
installed_review: InstalledEvidenceReview,
|
|
evidence: dict[str, Any] | None,
|
|
evidence_schema: dict[str, Any] | None,
|
|
authority_keyring: dict[str, Any] | None,
|
|
authority_keyring_schema: dict[str, Any] | None,
|
|
verification_time: datetime | None = None,
|
|
verification_mode: str = "current",
|
|
expected_external_provider_subject: str | None = None,
|
|
forbidden_authority_public_keys: frozenset[bytes] = frozenset(),
|
|
) -> BoundaryEvidenceReview:
|
|
deployment_profile = assessment.get("deployment_profile")
|
|
raw_deployment_subject = (
|
|
deployment_profile.get("id") if isinstance(deployment_profile, dict) else None
|
|
)
|
|
deployment_subject = _safe_opaque_id_or_none(raw_deployment_subject)
|
|
provider_subject = _safe_opaque_id_or_none(expected_external_provider_subject)
|
|
effective_verification_mode = (
|
|
verification_mode if verification_mode in VERIFICATION_MODES else "invalid"
|
|
)
|
|
expected_subjects = {
|
|
"target_environment": deployment_subject,
|
|
"external_providers": provider_subject,
|
|
"production_approval": deployment_subject,
|
|
}
|
|
scopes = _unchecked_boundary_scope(expected_subjects=expected_subjects)
|
|
findings: list[EvidenceFinding] = []
|
|
if raw_deployment_subject is not None and deployment_subject is None:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_deployment_subject_invalid",
|
|
"The assessment deployment profile has no valid opaque subject ID.",
|
|
("assessment.deployment_profile",),
|
|
)
|
|
)
|
|
if expected_external_provider_subject is not None and provider_subject is None:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_provider_subject_invalid",
|
|
"The expected external-provider subject must be a bounded opaque ID.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
if effective_verification_mode == "invalid":
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_verification_mode_invalid",
|
|
"Boundary evidence verification mode is invalid.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
if evidence is None:
|
|
return BoundaryEvidenceReview(findings=tuple(findings), proof_scope=scopes)
|
|
now = verification_time or datetime.now(tz=UTC)
|
|
if now.tzinfo is None:
|
|
now = now.replace(tzinfo=UTC)
|
|
scopes["boundary_evidence_verification"] = {
|
|
"checked": True,
|
|
"evaluated_at": _iso_datetime(now),
|
|
"verification_mode": effective_verification_mode,
|
|
"historical": effective_verification_mode == "historical_override",
|
|
}
|
|
if evidence_schema is None or authority_keyring_schema is None:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_evidence_schema_missing",
|
|
"Boundary evidence was supplied without both validation schemas.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
return BoundaryEvidenceReview(tuple(findings), scopes)
|
|
evidence_errors = validate_payload(payload=evidence, schema=evidence_schema)
|
|
keyring_errors = (
|
|
validate_payload(payload=authority_keyring, schema=authority_keyring_schema)
|
|
if isinstance(authority_keyring, dict)
|
|
else ("$: an independently provisioned proof-authority keyring is required",)
|
|
)
|
|
findings.extend(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_evidence_schema",
|
|
message,
|
|
("assessment.external_proof",),
|
|
)
|
|
for message in evidence_errors
|
|
)
|
|
findings.extend(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_authority_keyring_schema",
|
|
message,
|
|
("assessment.external_proof",),
|
|
)
|
|
for message in keyring_errors
|
|
)
|
|
if evidence_errors or keyring_errors or not isinstance(authority_keyring, dict):
|
|
return BoundaryEvidenceReview(tuple(findings), scopes)
|
|
|
|
expected_release = (
|
|
assessment.get("release", {}).get("ref")
|
|
if isinstance(assessment.get("release"), dict)
|
|
else None
|
|
)
|
|
if evidence.get("assessment_id") != assessment.get("assessment_id"):
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_assessment_mismatch",
|
|
"Boundary evidence is bound to a different assessment ID.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
if evidence.get("assessment_release") != expected_release:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_release_mismatch",
|
|
"Boundary evidence is bound to a different assessment release.",
|
|
("assessment.release", "assessment.external_proof"),
|
|
)
|
|
)
|
|
installed_digest = installed_review.evidence_sha256
|
|
installed_scope = installed_review.proof_scope.get("installed_artifacts", {})
|
|
if (
|
|
installed_digest is None
|
|
or evidence.get("installed_evidence_sha256") != installed_digest
|
|
or installed_scope.get("valid") is not True
|
|
):
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_installed_evidence_untrusted",
|
|
"Boundary evidence must bind to the supplied, valid installed-composition evidence digest.",
|
|
("assessment.installed_composition", "assessment.external_proof"),
|
|
)
|
|
)
|
|
|
|
issued_at = _datetime(evidence.get("issued_at"))
|
|
expires_at = _datetime(evidence.get("expires_at"))
|
|
time_valid = True
|
|
if issued_at is None or expires_at is None or issued_at >= expires_at:
|
|
time_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_evidence_interval_invalid",
|
|
"Boundary evidence must have issued_at earlier than expires_at.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
elif now < issued_at or now >= expires_at:
|
|
time_valid = False
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_evidence_not_current",
|
|
"Boundary evidence is not valid at the verification time.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
|
|
claims = [item for item in evidence.get("claims", []) if isinstance(item, dict)]
|
|
claims_by_scope: dict[str, dict[str, Any]] = {}
|
|
duplicate_scopes: set[str] = set()
|
|
semantic_invalid: set[str] = set()
|
|
subject_invalid: set[str] = set()
|
|
for claim in claims:
|
|
scope = str(claim.get("scope") or "")
|
|
if scope in claims_by_scope:
|
|
duplicate_scopes.add(scope)
|
|
else:
|
|
claims_by_scope[scope] = claim
|
|
allowed_results = {POSITIVE_RESULTS.get(scope), NEGATIVE_RESULTS.get(scope)}
|
|
if claim.get("result") not in allowed_results:
|
|
semantic_invalid.add(scope)
|
|
controls = claim.get("control_ids")
|
|
if not isinstance(controls, list) or len(controls) != len(set(controls)):
|
|
semantic_invalid.add(scope)
|
|
artifacts = claim.get("artifacts")
|
|
artifact_ids = [
|
|
str(item.get("artifact_id") or "")
|
|
for item in artifacts or []
|
|
if isinstance(item, dict)
|
|
]
|
|
if len(artifact_ids) != len(set(artifact_ids)):
|
|
semantic_invalid.add(scope)
|
|
observed_subject = str(claim.get("subject_id") or "")
|
|
expected_subject = expected_subjects.get(scope)
|
|
if scope in scopes:
|
|
scopes[scope]["observed_subject_id"] = observed_subject
|
|
if expected_subject is None or observed_subject != expected_subject:
|
|
subject_invalid.add(scope)
|
|
for scope in sorted(duplicate_scopes):
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_claim_duplicate",
|
|
f"Boundary evidence repeats claim scope {scope!r}.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
for scope in sorted(semantic_invalid):
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_claim_semantics",
|
|
f"Boundary claim {scope!r} has an invalid result or duplicate control/artifact IDs.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
for scope in sorted(subject_invalid):
|
|
expected_subject = expected_subjects.get(scope)
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
(
|
|
"boundary_provider_subject_unconfigured"
|
|
if scope == "external_providers" and expected_subject is None
|
|
else "boundary_claim_subject_mismatch"
|
|
),
|
|
(
|
|
"External-provider proof requires an explicit expected opaque subject ID."
|
|
if scope == "external_providers" and expected_subject is None
|
|
else f"Boundary claim {scope!r} is not bound to its expected assessment subject."
|
|
),
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
|
|
authority_keys, key_findings = _active_authority_keys(
|
|
authority_keyring=authority_keyring,
|
|
verification_time=now,
|
|
forbidden_public_keys=forbidden_authority_public_keys,
|
|
)
|
|
findings.extend(key_findings)
|
|
signature_payload = dict(evidence)
|
|
signature_payload.pop("signatures", None)
|
|
signature_bytes = canonical_bytes(signature_payload)
|
|
authorized_scopes: set[str] = set()
|
|
seen_signature_key_ids: set[str] = set()
|
|
duplicate_signature_key_ids: set[str] = set()
|
|
for signature in evidence.get("signatures", []):
|
|
if not isinstance(signature, dict):
|
|
continue
|
|
key_id = str(signature.get("key_id") or "")
|
|
if key_id in seen_signature_key_ids:
|
|
duplicate_signature_key_ids.add(key_id)
|
|
continue
|
|
seen_signature_key_ids.add(key_id)
|
|
key = authority_keys.get(key_id)
|
|
if key is None or signature.get("algorithm") != "ed25519":
|
|
continue
|
|
try:
|
|
signature_value = base64.b64decode(
|
|
str(signature.get("value") or ""), validate=True
|
|
)
|
|
Ed25519PublicKey.from_public_bytes(key[0]).verify(
|
|
signature_value, signature_bytes
|
|
)
|
|
except (ValueError, InvalidSignature):
|
|
continue
|
|
authorized_scopes.update(key[1])
|
|
if duplicate_signature_key_ids:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_signature_duplicate",
|
|
"Boundary evidence repeats one or more signature key IDs.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
|
|
unauthorized_scopes = {
|
|
scope
|
|
for scope in claims_by_scope
|
|
if scope in PROOF_REQUIREMENTS and scope not in authorized_scopes
|
|
}
|
|
for scope in sorted(unauthorized_scopes):
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_authority_untrusted",
|
|
f"No valid independently trusted signature is authorized for {scope!r}.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
global_valid = not any(item.severity == "blocker" for item in findings)
|
|
evidence_digest = canonical_sha256(evidence)
|
|
review_targets: list[dict[str, Any]] = []
|
|
for scope, claim in sorted(claims_by_scope.items()):
|
|
if scope not in PROOF_REQUIREMENTS:
|
|
continue
|
|
if scope in unauthorized_scopes:
|
|
continue
|
|
if (
|
|
scope in duplicate_scopes
|
|
or scope in semantic_invalid
|
|
or scope in subject_invalid
|
|
or not time_valid
|
|
):
|
|
continue
|
|
if not global_valid:
|
|
continue
|
|
result = str(claim.get("result"))
|
|
valid = result == POSITIVE_RESULTS[scope]
|
|
scopes[scope] = {
|
|
"checked": True,
|
|
"valid": valid,
|
|
"evidence_sha256": evidence_digest,
|
|
"result": result,
|
|
"control_count": len(claim.get("control_ids") or []),
|
|
"artifact_count": len(claim.get("artifacts") or []),
|
|
"expected_subject_id": expected_subjects[scope],
|
|
"observed_subject_id": claim.get("subject_id"),
|
|
"evaluated_at": _iso_datetime(now),
|
|
"verification_mode": effective_verification_mode,
|
|
"required_evidence": PROOF_REQUIREMENTS[scope],
|
|
}
|
|
if not valid:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"review",
|
|
"boundary_claim_negative",
|
|
f"Authorized boundary claim {scope!r} has result {result!r}.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
subject = str(expected_subjects[scope])
|
|
review_targets.append(
|
|
{
|
|
"id": f"boundary.{scope}.{subject}",
|
|
"section": scope,
|
|
"reason": f"Authorized boundary evidence reported {result!r} for {scope!r}.",
|
|
}
|
|
)
|
|
return BoundaryEvidenceReview(
|
|
tuple(findings),
|
|
scopes,
|
|
tuple(
|
|
{
|
|
item["id"]: item
|
|
for item in sorted(review_targets, key=lambda value: value["id"])
|
|
}.values()
|
|
),
|
|
)
|
|
|
|
|
|
def normalize_package_name(value: str) -> str:
|
|
return re.sub(r"[-_.]+", "-", value.strip().lower())
|
|
|
|
|
|
def canonical_bytes(payload: object) -> bytes:
|
|
return json.dumps(
|
|
payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode("utf-8")
|
|
|
|
|
|
def canonical_sha256(payload: object) -> str:
|
|
return hashlib.sha256(canonical_bytes(payload)).hexdigest()
|
|
|
|
|
|
def public_key_material_from_keyrings(
|
|
*payloads: Mapping[str, Any],
|
|
) -> frozenset[bytes]:
|
|
result: set[bytes] = set()
|
|
for payload in payloads:
|
|
keys = payload.get("keys")
|
|
values = (
|
|
[item.get("public_key") for item in keys if isinstance(item, dict)]
|
|
if isinstance(keys, list)
|
|
else list(payload.values())
|
|
)
|
|
for value in values:
|
|
if not isinstance(value, str):
|
|
continue
|
|
try:
|
|
decoded = base64.b64decode(value, validate=True)
|
|
Ed25519PublicKey.from_public_bytes(decoded)
|
|
except ValueError:
|
|
continue
|
|
result.add(decoded)
|
|
return frozenset(result)
|
|
|
|
|
|
def _distribution_version(distribution: metadata.Distribution) -> str:
|
|
try:
|
|
value = str(distribution.version)
|
|
except Exception:
|
|
return ""
|
|
if len(value) > 128 or VERSION_PATTERN.fullmatch(value) is None:
|
|
return ""
|
|
return value
|
|
|
|
|
|
def _distribution_modules(
|
|
*, package_name: str, distribution: metadata.Distribution
|
|
) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
|
|
modules: list[dict[str, str]] = []
|
|
issues: list[dict[str, str]] = []
|
|
try:
|
|
all_entry_points = []
|
|
for entry_point in distribution.entry_points:
|
|
if entry_point.group != "govoplan.modules":
|
|
continue
|
|
name = str(entry_point.name)
|
|
value = str(entry_point.value)
|
|
if (
|
|
len(name) > 160
|
|
or OPAQUE_ID_PATTERN.fullmatch(name) is None
|
|
or len(value) > 512
|
|
or ENTRY_POINT_VALUE_PATTERN.fullmatch(value) is None
|
|
):
|
|
issues.append(
|
|
{
|
|
"code": "module-entry-point-invalid",
|
|
"package_name": package_name,
|
|
}
|
|
)
|
|
continue
|
|
all_entry_points.append((name, value, entry_point))
|
|
all_entry_points.sort(key=lambda item: (item[0], item[1]))
|
|
except Exception:
|
|
return [], [
|
|
{
|
|
"code": "distribution-metadata-invalid",
|
|
"package_name": package_name,
|
|
}
|
|
]
|
|
if len(all_entry_points) > 16:
|
|
issues.append(
|
|
{
|
|
"code": "module-entry-point-limit-exceeded",
|
|
"package_name": package_name,
|
|
}
|
|
)
|
|
for _, _, entry_point in all_entry_points[:16]:
|
|
try:
|
|
manifest = entry_point.load()()
|
|
module_id = str(manifest.id)
|
|
manifest_version = str(manifest.version)
|
|
except (Exception, SystemExit):
|
|
issues.append(
|
|
{
|
|
"code": "module-entry-point-load-failed",
|
|
"package_name": package_name,
|
|
}
|
|
)
|
|
continue
|
|
if (
|
|
len(module_id) > 160
|
|
or OPAQUE_ID_PATTERN.fullmatch(module_id) is None
|
|
or len(manifest_version) > 128
|
|
or VERSION_PATTERN.fullmatch(manifest_version) is None
|
|
):
|
|
issues.append(
|
|
{
|
|
"code": "module-entry-point-invalid",
|
|
"package_name": package_name,
|
|
}
|
|
)
|
|
continue
|
|
modules.append({"module_id": module_id, "manifest_version": manifest_version})
|
|
return sorted(modules, key=lambda item: item["module_id"]), issues
|
|
|
|
|
|
def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
|
|
try:
|
|
raw = distribution.read_text("direct_url.json")
|
|
except Exception:
|
|
return {"kind": "malformed-direct-url"}
|
|
if raw is None:
|
|
return {"kind": "index-or-unknown"}
|
|
if not isinstance(raw, str) or len(raw) > MAX_DIRECT_URL_BYTES:
|
|
return {"kind": "malformed-direct-url"}
|
|
if len(raw.encode("utf-8")) > MAX_DIRECT_URL_BYTES:
|
|
return {"kind": "malformed-direct-url"}
|
|
try:
|
|
direct_url = json.loads(raw)
|
|
except (TypeError, json.JSONDecodeError):
|
|
return {"kind": "malformed-direct-url"}
|
|
if not isinstance(direct_url, dict):
|
|
return {"kind": "malformed-direct-url"}
|
|
raw_url = direct_url.get("url")
|
|
if (
|
|
not isinstance(raw_url, str)
|
|
or not raw_url
|
|
or len(raw_url) > 2048
|
|
or any(character.isspace() or ord(character) < 32 for character in raw_url)
|
|
):
|
|
return {"kind": "malformed-direct-url"}
|
|
try:
|
|
parsed_url = urlsplit(raw_url)
|
|
except ValueError:
|
|
return {"kind": "malformed-direct-url"}
|
|
if parsed_url.password is not None or parsed_url.query or parsed_url.fragment:
|
|
return {"kind": "malformed-direct-url"}
|
|
vcs_info = direct_url.get("vcs_info")
|
|
directory_info = direct_url.get("dir_info")
|
|
archive_info = direct_url.get("archive_info")
|
|
provenance_forms = tuple(
|
|
name
|
|
for name, value in (
|
|
("vcs", vcs_info),
|
|
("directory", directory_info),
|
|
("archive", archive_info),
|
|
)
|
|
if value is not None
|
|
)
|
|
if len(provenance_forms) != 1:
|
|
return {"kind": "malformed-direct-url"}
|
|
if isinstance(vcs_info, dict):
|
|
if vcs_info.get("vcs") != "git":
|
|
return {"kind": "malformed-direct-url"}
|
|
commit = str(vcs_info.get("commit_id") or "").lower()
|
|
requested_revision = vcs_info.get("requested_revision")
|
|
if (
|
|
GIT_OBJECT_PATTERN.fullmatch(commit) is None
|
|
or parsed_url.scheme
|
|
not in {"file", "git", "git+https", "git+ssh", "https", "ssh"}
|
|
or not parsed_url.path
|
|
or (
|
|
parsed_url.scheme != "file"
|
|
and (not parsed_url.hostname or not parsed_url.path.endswith(".git"))
|
|
)
|
|
or (parsed_url.scheme in {"https", "git+https"} and parsed_url.username)
|
|
or (
|
|
requested_revision is not None
|
|
and (
|
|
not isinstance(requested_revision, str)
|
|
or VCS_REVISION_PATTERN.fullmatch(requested_revision) is None
|
|
)
|
|
)
|
|
):
|
|
return {"kind": "malformed-direct-url"}
|
|
return {"kind": "vcs-commit", "commit": commit}
|
|
if isinstance(archive_info, dict):
|
|
if (
|
|
parsed_url.scheme not in {"file", "https"}
|
|
or not parsed_url.path
|
|
or (
|
|
parsed_url.scheme == "https"
|
|
and (not parsed_url.hostname or parsed_url.username)
|
|
)
|
|
):
|
|
return {"kind": "malformed-direct-url"}
|
|
hashes = archive_info.get("hashes")
|
|
digest = hashes.get("sha256") if isinstance(hashes, dict) else None
|
|
if digest is None:
|
|
legacy = str(archive_info.get("hash") or "")
|
|
if legacy.startswith("sha256="):
|
|
digest = legacy.removeprefix("sha256=")
|
|
digest_text = str(digest or "").lower()
|
|
if SHA256_PATTERN.fullmatch(digest_text):
|
|
return {"kind": "archive", "sha256": digest_text}
|
|
return {"kind": "malformed-direct-url"}
|
|
if isinstance(directory_info, dict):
|
|
editable = directory_info.get("editable", False)
|
|
if (
|
|
parsed_url.scheme != "file"
|
|
or not parsed_url.path
|
|
or not isinstance(editable, bool)
|
|
):
|
|
return {"kind": "malformed-direct-url"}
|
|
return {"kind": "editable-local" if editable else "local-directory"}
|
|
return {"kind": "malformed-direct-url"}
|
|
|
|
|
|
def _record_integrity(
|
|
distribution: metadata.Distribution,
|
|
*,
|
|
package_name: str,
|
|
budget: CollectionHashBudget,
|
|
) -> dict[str, Any]:
|
|
counts = {
|
|
"hashed_file_count": 0,
|
|
"permitted_unhashed_file_count": 0,
|
|
"generated_unhashed_file_count": 0,
|
|
"unverifiable_file_count": 0,
|
|
"missing_file_count": 0,
|
|
"mismatched_file_count": 0,
|
|
}
|
|
try:
|
|
files = tuple(distribution.files or ())
|
|
except Exception:
|
|
files = ()
|
|
if not files:
|
|
return {"status": "unavailable", **counts}
|
|
if len(files) > MAX_DISTRIBUTION_FILES or len(files) > budget.remaining_files:
|
|
budget.remaining_files = 0
|
|
return {"status": "limit-exceeded", **counts}
|
|
budget.remaining_files -= len(files)
|
|
try:
|
|
distribution_root = Path(distribution.locate_file("")).resolve()
|
|
installation_root = Path(sys.prefix).resolve()
|
|
except OSError:
|
|
return {"status": "unavailable", **counts}
|
|
if not _trusted_distribution_root(distribution_root):
|
|
return {"status": "unavailable", **counts}
|
|
declared_scripts = _declared_distribution_scripts(distribution)
|
|
owner_prefix = package_name.replace("-", "_")
|
|
resolved_entries: list[tuple[object, Path | None]] = [
|
|
(
|
|
package_path,
|
|
_resolve_record_path(
|
|
distribution=distribution,
|
|
package_path=package_path,
|
|
distribution_root=distribution_root,
|
|
installation_root=installation_root,
|
|
owner_prefix=owner_prefix,
|
|
declared_scripts=declared_scripts,
|
|
),
|
|
)
|
|
for package_path in files
|
|
]
|
|
hashed_payload_paths = {
|
|
resolved_path
|
|
for package_path, resolved_path in resolved_entries
|
|
if resolved_path is not None
|
|
and getattr(getattr(package_path, "hash", None), "mode", None) == "sha256"
|
|
}
|
|
total_hashed_bytes = 0
|
|
limit_exceeded = False
|
|
for package_path, resolved_path in resolved_entries:
|
|
hash_info = getattr(package_path, "hash", None)
|
|
if resolved_path is None:
|
|
counts["unverifiable_file_count"] += 1
|
|
continue
|
|
if hash_info is None:
|
|
if _permitted_unhashed_record_path(
|
|
resolved_path,
|
|
distribution_root=distribution_root,
|
|
owner_prefix=owner_prefix,
|
|
):
|
|
counts["permitted_unhashed_file_count"] += 1
|
|
elif _generated_unhashed_pyc(
|
|
resolved_path,
|
|
hashed_payload_paths=hashed_payload_paths,
|
|
):
|
|
counts["generated_unhashed_file_count"] += 1
|
|
else:
|
|
counts["unverifiable_file_count"] += 1
|
|
continue
|
|
if getattr(hash_info, "mode", None) != "sha256":
|
|
counts["unverifiable_file_count"] += 1
|
|
continue
|
|
try:
|
|
file_stat = resolved_path.stat()
|
|
except OSError:
|
|
counts["missing_file_count"] += 1
|
|
continue
|
|
if not stat.S_ISREG(file_stat.st_mode):
|
|
counts["missing_file_count"] += 1
|
|
continue
|
|
remaining_distribution_bytes = (
|
|
MAX_HASHED_DISTRIBUTION_BYTES - total_hashed_bytes
|
|
)
|
|
allowed_bytes = min(
|
|
MAX_HASHED_FILE_BYTES,
|
|
remaining_distribution_bytes,
|
|
budget.remaining_bytes,
|
|
)
|
|
if allowed_bytes <= 0 or file_stat.st_size > allowed_bytes:
|
|
limit_exceeded = True
|
|
continue
|
|
expected = _record_hash_bytes(str(getattr(hash_info, "value", "")))
|
|
if expected is None:
|
|
counts["unverifiable_file_count"] += 1
|
|
continue
|
|
try:
|
|
actual, hashed_bytes, exceeded = _bounded_file_sha256(
|
|
resolved_path,
|
|
max_bytes=allowed_bytes,
|
|
)
|
|
except OSError:
|
|
counts["missing_file_count"] += 1
|
|
continue
|
|
total_hashed_bytes += hashed_bytes
|
|
budget.remaining_bytes -= hashed_bytes
|
|
if exceeded:
|
|
limit_exceeded = True
|
|
continue
|
|
if hmac.compare_digest(actual, expected):
|
|
counts["hashed_file_count"] += 1
|
|
else:
|
|
counts["mismatched_file_count"] += 1
|
|
if limit_exceeded:
|
|
status = "limit-exceeded"
|
|
elif counts["missing_file_count"] or counts["mismatched_file_count"]:
|
|
status = "mismatch"
|
|
elif counts["hashed_file_count"] == 0:
|
|
status = "unavailable"
|
|
elif counts["unverifiable_file_count"]:
|
|
status = "partial"
|
|
else:
|
|
status = "verified"
|
|
return {"status": status, **counts}
|
|
|
|
|
|
def _resolve_record_path(
|
|
*,
|
|
distribution: metadata.Distribution,
|
|
package_path: object,
|
|
distribution_root: Path,
|
|
installation_root: Path,
|
|
owner_prefix: str,
|
|
declared_scripts: frozenset[str],
|
|
) -> Path | None:
|
|
record_path = PurePosixPath(str(package_path).replace("\\", "/"))
|
|
if record_path.is_absolute() or not record_path.parts:
|
|
return None
|
|
try:
|
|
unresolved_path = Path(distribution.locate_file(package_path))
|
|
if unresolved_path.is_symlink():
|
|
return None
|
|
resolved_path = unresolved_path.resolve()
|
|
except OSError:
|
|
return None
|
|
has_parent_traversal = ".." in record_path.parts
|
|
if not has_parent_traversal:
|
|
return (
|
|
resolved_path if _is_relative_to(resolved_path, distribution_root) else None
|
|
)
|
|
script_roots = (
|
|
(installation_root / "bin").resolve(),
|
|
(installation_root / "Scripts").resolve(),
|
|
)
|
|
for script_root in script_roots:
|
|
if (
|
|
resolved_path.parent == script_root
|
|
and resolved_path.name in declared_scripts
|
|
):
|
|
return resolved_path
|
|
if not _is_relative_to(resolved_path, installation_root):
|
|
return None
|
|
relative = resolved_path.relative_to(installation_root)
|
|
if not relative.parts:
|
|
return None
|
|
owned_root = relative.parts[0]
|
|
if owned_root == owner_prefix or owned_root.startswith(f"{owner_prefix}_"):
|
|
return resolved_path
|
|
return None
|
|
|
|
|
|
def _declared_distribution_scripts(
|
|
distribution: metadata.Distribution,
|
|
) -> frozenset[str]:
|
|
scripts: set[str] = set()
|
|
try:
|
|
entry_points = distribution.entry_points
|
|
except Exception:
|
|
return frozenset()
|
|
for entry_point in entry_points:
|
|
if entry_point.group != "console_scripts":
|
|
continue
|
|
name = str(entry_point.name)
|
|
if len(name) > 160 or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", name) is None:
|
|
continue
|
|
scripts.update((name, f"{name}.exe", f"{name}-script.py"))
|
|
return frozenset(scripts)
|
|
|
|
|
|
def _bounded_file_sha256(path: Path, *, max_bytes: int) -> tuple[bytes, int, bool]:
|
|
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
|
descriptor = os.open(path, flags)
|
|
digest = hashlib.sha256()
|
|
total = 0
|
|
exceeded = False
|
|
try:
|
|
file_stat = os.fstat(descriptor)
|
|
if not stat.S_ISREG(file_stat.st_mode):
|
|
raise OSError("RECORD path is not a regular file")
|
|
if file_stat.st_size > max_bytes:
|
|
return digest.digest(), 0, True
|
|
expected_size = file_stat.st_size
|
|
with os.fdopen(descriptor, "rb", closefd=False) as handle:
|
|
while total < expected_size:
|
|
chunk = handle.read(min(1024 * 1024, expected_size - total))
|
|
if not chunk:
|
|
break
|
|
digest.update(chunk)
|
|
total += len(chunk)
|
|
final_stat = os.fstat(descriptor)
|
|
exceeded = total != expected_size or final_stat.st_size != expected_size
|
|
finally:
|
|
os.close(descriptor)
|
|
return digest.digest(), total, exceeded
|
|
|
|
|
|
def _record_hash_bytes(value: str) -> bytes | None:
|
|
try:
|
|
decoded = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
|
|
except (ValueError, TypeError):
|
|
return None
|
|
return decoded if len(decoded) == hashlib.sha256().digest_size else None
|
|
|
|
|
|
def _record_integrity_semantics_valid(record: Mapping[str, Any]) -> bool:
|
|
count_names = (
|
|
"hashed_file_count",
|
|
"permitted_unhashed_file_count",
|
|
"generated_unhashed_file_count",
|
|
"unverifiable_file_count",
|
|
"missing_file_count",
|
|
"mismatched_file_count",
|
|
)
|
|
counts: dict[str, int] = {}
|
|
for name in count_names:
|
|
value = record.get(name)
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
return False
|
|
counts[name] = value
|
|
if sum(counts.values()) > MAX_DISTRIBUTION_FILES:
|
|
return False
|
|
status = record.get("status")
|
|
hashed = counts["hashed_file_count"]
|
|
unverifiable = counts["unverifiable_file_count"]
|
|
missing = counts["missing_file_count"]
|
|
mismatched = counts["mismatched_file_count"]
|
|
if status == "verified":
|
|
return (
|
|
hashed > 0
|
|
and counts["permitted_unhashed_file_count"] > 0
|
|
and unverifiable == 0
|
|
and missing == 0
|
|
and mismatched == 0
|
|
)
|
|
if status == "partial":
|
|
return hashed > 0 and unverifiable > 0 and missing == 0 and mismatched == 0
|
|
if status == "mismatch":
|
|
return missing + mismatched > 0
|
|
if status == "unavailable":
|
|
return hashed == 0 and missing == 0 and mismatched == 0
|
|
if status == "limit-exceeded":
|
|
return True
|
|
return False
|
|
|
|
|
|
def _permitted_unhashed_record_path(
|
|
path: Path,
|
|
*,
|
|
distribution_root: Path,
|
|
owner_prefix: str,
|
|
) -> bool:
|
|
if not _is_relative_to(path, distribution_root):
|
|
return False
|
|
if path.name not in {"RECORD", "RECORD.jws", "RECORD.p7s"}:
|
|
return False
|
|
parent = path.parent.name
|
|
return parent.endswith(".dist-info") and (
|
|
parent == f"{owner_prefix}.dist-info" or parent.startswith(f"{owner_prefix}-")
|
|
)
|
|
|
|
|
|
def _generated_unhashed_pyc(
|
|
path: Path,
|
|
*,
|
|
hashed_payload_paths: set[Path],
|
|
) -> bool:
|
|
if path.suffix != ".pyc" or "__pycache__" not in path.parts:
|
|
return False
|
|
try:
|
|
source = Path(source_from_cache(str(path))).resolve()
|
|
except (ValueError, OSError):
|
|
return False
|
|
return source in hashed_payload_paths
|
|
|
|
|
|
def _active_authority_keys(
|
|
*,
|
|
authority_keyring: dict[str, Any],
|
|
verification_time: datetime,
|
|
forbidden_public_keys: frozenset[bytes],
|
|
) -> tuple[dict[str, tuple[bytes, frozenset[str]]], tuple[EvidenceFinding, ...]]:
|
|
result: dict[str, tuple[bytes, frozenset[str]]] = {}
|
|
findings: list[EvidenceFinding] = []
|
|
seen: set[str] = set()
|
|
public_key_owners: dict[bytes, str] = {}
|
|
duplicate_public_keys: set[bytes] = set()
|
|
for item in authority_keyring.get("keys", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
key_id = str(item.get("key_id") or "")
|
|
if key_id in seen:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_authority_key_duplicate",
|
|
"Proof-authority keyring repeats a key ID.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
continue
|
|
seen.add(key_id)
|
|
try:
|
|
public_key = base64.b64decode(
|
|
str(item.get("public_key") or ""), validate=True
|
|
)
|
|
Ed25519PublicKey.from_public_bytes(public_key)
|
|
except ValueError:
|
|
continue
|
|
previous_owner = public_key_owners.get(public_key)
|
|
if previous_owner is not None and previous_owner != key_id:
|
|
if public_key not in duplicate_public_keys:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_authority_public_key_duplicate",
|
|
"Proof-authority keyring assigns one public key to multiple key IDs.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
duplicate_public_keys.add(public_key)
|
|
result.pop(previous_owner, None)
|
|
continue
|
|
public_key_owners[public_key] = key_id
|
|
if item.get("status") not in {"active", "next"}:
|
|
continue
|
|
not_before = _datetime(item.get("not_before"))
|
|
not_after = _datetime(item.get("not_after"))
|
|
if (not_before is not None and verification_time < not_before) or (
|
|
not_after is not None and verification_time >= not_after
|
|
):
|
|
continue
|
|
if public_key in forbidden_public_keys:
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"blocker",
|
|
"boundary_authority_reuses_release_key",
|
|
"A proof-authority public key is also trusted for catalog release signatures.",
|
|
("assessment.external_proof",),
|
|
)
|
|
)
|
|
continue
|
|
result[key_id] = (
|
|
public_key,
|
|
frozenset(str(value) for value in item.get("allowed_scopes", [])),
|
|
)
|
|
return result, tuple(findings)
|
|
|
|
|
|
def _record_change(
|
|
*,
|
|
findings: list[EvidenceFinding],
|
|
changes: list[dict[str, Any]],
|
|
changed_repositories: set[str],
|
|
affected_module_ids: set[str],
|
|
component: Mapping[str, Any],
|
|
kind: str,
|
|
code: str,
|
|
message: str,
|
|
extra: Mapping[str, Any] | None = None,
|
|
) -> None:
|
|
module_id = str(component.get("module_id"))
|
|
repository = str(component.get("repository"))
|
|
changed_repositories.add(repository)
|
|
affected_module_ids.add(module_id)
|
|
findings.append(
|
|
EvidenceFinding(
|
|
"review",
|
|
code,
|
|
message,
|
|
(f"composition.{module_id}",),
|
|
)
|
|
)
|
|
change: dict[str, Any] = {
|
|
"module_id": module_id,
|
|
"repository": repository,
|
|
"kind": kind,
|
|
}
|
|
if extra:
|
|
change.update(extra)
|
|
changes.append(change)
|
|
|
|
|
|
def _unchecked_installed_scope() -> dict[str, Any]:
|
|
return {
|
|
"installed_artifacts": {
|
|
"checked": False,
|
|
"valid": None,
|
|
"required_evidence": (
|
|
"A bounded installed-composition evidence document collected from the Python environment under review.",
|
|
),
|
|
},
|
|
"installed_record_integrity": {"checked": False, "valid": None},
|
|
"installed_source_provenance": {"checked": False, "valid": None},
|
|
"runtime_activation": {
|
|
"checked": False,
|
|
"valid": None,
|
|
"required_evidence": (
|
|
"A separately collected runtime registry/configuration snapshot bound to installed evidence.",
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
def _invalid_installed_scope(*, digest: str) -> dict[str, Any]:
|
|
scope = _unchecked_installed_scope()
|
|
scope["installed_artifacts"] = {
|
|
"checked": False,
|
|
"valid": None,
|
|
"comparison_valid": False,
|
|
"evidence_sha256": digest,
|
|
}
|
|
return scope
|
|
|
|
|
|
def _unchecked_boundary_scope(
|
|
*, expected_subjects: Mapping[str, str | None]
|
|
) -> dict[str, Any]:
|
|
return {
|
|
scope: {
|
|
"checked": False,
|
|
"valid": None,
|
|
"expected_subject_id": expected_subjects.get(scope),
|
|
"required_evidence": requirements,
|
|
}
|
|
for scope, requirements in PROOF_REQUIREMENTS.items()
|
|
}
|
|
|
|
|
|
def _datetime(value: object) -> datetime | None:
|
|
if not isinstance(value, str) or not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
|
|
|
|
|
|
def _iso_datetime(value: datetime | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def _safe_opaque_id(value: object, *, fallback: str) -> str:
|
|
text = value if isinstance(value, str) else ""
|
|
return (
|
|
text
|
|
if len(text) <= 160 and OPAQUE_ID_PATTERN.fullmatch(text) is not None
|
|
else fallback
|
|
)
|
|
|
|
|
|
def _safe_opaque_id_or_none(value: object) -> str | None:
|
|
if not isinstance(value, str):
|
|
return None
|
|
if len(value) > 160 or OPAQUE_ID_PATTERN.fullmatch(value) is None:
|
|
return None
|
|
return value
|
|
|
|
|
|
def _json_path(parts: Iterable[object]) -> str:
|
|
path = "$"
|
|
for part in parts:
|
|
path += f"[{part}]" if isinstance(part, int) else f".{part}"
|
|
return path
|
|
|
|
|
|
def _unique_dicts(values: Sequence[dict[str, str]]) -> list[dict[str, str]]:
|
|
result: list[dict[str, str]] = []
|
|
seen: set[tuple[tuple[str, str], ...]] = set()
|
|
for value in values:
|
|
key = tuple(sorted(value.items()))
|
|
if key not in seen:
|
|
seen.add(key)
|
|
result.append(value)
|
|
return result
|
|
|
|
|
|
def _is_relative_to(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.relative_to(root)
|
|
except ValueError:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _trusted_distribution_root(root: Path) -> bool:
|
|
for value in sys.path:
|
|
try:
|
|
candidate = Path(value or ".").resolve()
|
|
except OSError:
|
|
continue
|
|
if candidate.parent == candidate:
|
|
continue
|
|
if root == candidate or _is_relative_to(root, candidate):
|
|
return True
|
|
return False
|