yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: bind license materials to detected components

Browse Source
This commit is contained in:
yuqei
2026-09-27 02:33:08 +08:00
parent e2de02ea5e
commit 2e58f4d8c0
2 changed files with 729 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuqei-infra/scripts/build-third-party-materials.py
+533 -58
View File
@@ -29,8 +29,8 @@ from typing import Any, Iterable, Mapping, Sequence
INPUT_SCHEMA = "yuqei.third-party-materials-input.v1"
OUTPUT_SCHEMA = "yuqei.third-party-materials.v1"
COVERAGE_SCHEMA = "yuqei.third-party-occurrence-coverage.v1"
OFFER_SCHEMA = "yuqei.written-source-offer.v1"
VERIFY_SCHEMA = "yuqei.third-party-materials-verification.v1"
SOURCE_ARCHIVE_SCHEMA = "yuqei.corresponding-source-archive.v1"
PROFILES = {"internal-hosted", "redistributable"}
EXPECTED_ROLES = {
"contract-web",
@@ -54,6 +54,8 @@ MAX_LICENSE_TEXT = 8 * 1024 * 1024
MAX_ARCHIVE_FILES = 200_000
MAX_ARCHIVE_EXPANDED = 8 * 1024 * 1024 * 1024
MAX_COMPRESSION_RATIO = 2_000
SOURCE_MANIFEST_NAME = "YUQEI-SOURCE-MANIFEST.json"
SPDX_TOKEN = re.compile(r"[A-Za-z0-9][A-Za-z0-9.+-]*")
class MaterialsError(RuntimeError):
@@ -198,7 +200,7 @@ def _file_record(root: Path, raw: Any, label: str) -> tuple[dict[str, Any], Path
return {"path": relative, "sha256": actual_hash, "size_bytes": actual_size}, path
def _load_json(path: Path, label: str) -> Mapping[str, Any]:
def _loads_json_strict(raw: str, label: str) -> Mapping[str, Any]:
def no_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
@@ -209,17 +211,25 @@ def _load_json(path: Path, label: str) -> Mapping[str, Any]:
try:
value = json.loads(
path.read_text(encoding="utf-8"),
raw,
object_pairs_hook=no_duplicate_pairs,
parse_constant=lambda value: (_ for _ in ()).throw(
MaterialsError(f"non-finite JSON value in {label}: {value}")
),
)
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise MaterialsError(f"invalid JSON for {label}: {path}") from exc
except json.JSONDecodeError as exc:
raise MaterialsError(f"invalid JSON for {label}") from exc
return _mapping(value, label)
def _load_json(path: Path, label: str) -> Mapping[str, Any]:
try:
raw = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
raise MaterialsError(f"invalid JSON for {label}: {path}") from exc
return _loads_json_strict(raw, label)
def _parse_time(value: Any, label: str) -> datetime:
text = _text(value, label)
if text.endswith("Z"):
@@ -233,10 +243,174 @@ def _parse_time(value: Any, label: str) -> datetime:
return parsed.astimezone(timezone.utc)
def _validate_archive(path: Path, label: str) -> dict[str, int]:
def _spdx_tokens(value: Any, label: str) -> list[str]:
expression = _text(value, label)
tokens: list[str] = []
position = 0
while position < len(expression):
while position < len(expression) and expression[position].isspace():
position += 1
if position == len(expression):
break
if expression[position] in "()":
tokens.append(expression[position])
position += 1
continue
match = SPDX_TOKEN.match(expression, position)
if not match:
raise MaterialsError(f"invalid SPDX expression token in {label} at offset {position}")
tokens.append(match.group(0))
position = match.end()
if not tokens:
raise MaterialsError(f"empty SPDX expression for {label}")
return tokens
def _normalize_spdx_ast(ast: tuple[Any, ...]) -> tuple[Any, ...]:
operator = ast[0]
if operator not in {"and", "or"}:
return ast
children: list[tuple[Any, ...]] = []
for child in ast[1]:
normalized = _normalize_spdx_ast(child)
if normalized[0] == operator:
children.extend(normalized[1])
else:
children.append(normalized)
return (operator, tuple(sorted(children, key=repr)))
def _parse_spdx_expression(
value: Any,
label: str,
license_ids: set[str],
exception_ids: set[str],
) -> tuple[Any, ...]:
tokens = _spdx_tokens(value, label)
position = 0
def primary() -> tuple[Any, ...]:
nonlocal position
if position >= len(tokens):
raise MaterialsError(f"incomplete SPDX expression for {label}")
token = tokens[position]
if token == "(":
position += 1
result = parse_or()
if position >= len(tokens) or tokens[position] != ")":
raise MaterialsError(f"unbalanced SPDX expression for {label}")
position += 1
return result
if token in {"AND", "OR", "WITH", ")"} or token not in license_ids:
raise MaterialsError(f"unknown or non-canonical SPDX license ID in {label}: {token}")
position += 1
return ("id", token)
def parse_with() -> tuple[Any, ...]:
nonlocal position
result = primary()
if position < len(tokens) and tokens[position] == "WITH":
if result[0] != "id":
raise MaterialsError(f"SPDX WITH must follow one license ID in {label}")
position += 1
if position >= len(tokens) or tokens[position] not in exception_ids:
bad = tokens[position] if position < len(tokens) else "<missing>"
raise MaterialsError(f"unknown SPDX exception in {label}: {bad}")
result = ("with", result, tokens[position])
position += 1
return result
def parse_and() -> tuple[Any, ...]:
nonlocal position
children = [parse_with()]
while position < len(tokens) and tokens[position] == "AND":
position += 1
children.append(parse_with())
return children[0] if len(children) == 1 else ("and", tuple(children))
def parse_or() -> tuple[Any, ...]:
nonlocal position
children = [parse_and()]
while position < len(tokens) and tokens[position] == "OR":
position += 1
children.append(parse_and())
return children[0] if len(children) == 1 else ("or", tuple(children))
parsed = parse_or()
if position != len(tokens):
raise MaterialsError(f"unexpected SPDX token in {label}: {tokens[position]}")
return _normalize_spdx_ast(parsed)
def _spdx_terms(ast: tuple[Any, ...]) -> set[str]:
if ast[0] in {"id", "with"}:
return {_spdx_canonical(ast)}
terms: set[str] = set()
for child in ast[1]:
terms.update(_spdx_terms(child))
return terms
def _spdx_canonical(ast: tuple[Any, ...], parent_precedence: int = 0) -> str:
operator = ast[0]
if operator == "id":
return str(ast[1])
if operator == "with":
return f"{_spdx_canonical(ast[1], 3)} WITH {ast[2]}"
precedence = 2 if operator == "and" else 1
separator = " AND " if operator == "and" else " OR "
rendered = separator.join(_spdx_canonical(child, precedence) for child in ast[1])
return f"({rendered})" if precedence < parent_precedence else rendered
def _load_spdx_catalog(
root: Path, raw: Any
) -> tuple[set[str], set[str], list[dict[str, Any]]]:
catalog = _mapping(raw, "spdx_catalog")
if set(catalog) != {"licenses", "exceptions"}:
raise MaterialsError("spdx_catalog must contain exactly licenses and exceptions")
license_record, license_path = _file_record(root, catalog["licenses"], "SPDX license catalog")
exception_record, exception_path = _file_record(root, catalog["exceptions"], "SPDX exception catalog")
licenses_payload = _load_json(license_path, "SPDX license catalog")
exceptions_payload = _load_json(exception_path, "SPDX exception catalog")
if licenses_payload.get("licenseListVersion") != exceptions_payload.get("licenseListVersion"):
raise MaterialsError("SPDX license and exception catalog versions differ")
license_ids = [
_text(item.get("licenseId"), "SPDX license ID")
for item in (
_mapping(value, "SPDX license")
for value in _list(licenses_payload.get("licenses"), "SPDX licenses")
)
if not item.get("isDeprecatedLicenseId", False)
]
exception_ids = [
_text(item.get("licenseExceptionId"), "SPDX exception ID")
for item in (
_mapping(value, "SPDX exception")
for value in _list(exceptions_payload.get("exceptions"), "SPDX exceptions")
)
if not item.get("isDeprecatedLicenseId", False)
]
_unique(license_ids, "SPDX license ID")
_unique(exception_ids, "SPDX exception ID")
if not license_ids:
raise MaterialsError("SPDX license catalog is empty")
for identifier in [*license_ids, *exception_ids]:
if not SPDX_TOKEN.fullmatch(identifier) or "LicenseRef-" in identifier:
raise MaterialsError(f"invalid SPDX catalog identifier: {identifier}")
records = [
{"kind": "spdx-license-catalog", **license_record},
{"kind": "spdx-exception-catalog", **exception_record},
]
return set(license_ids), set(exception_ids), records
def _inspect_archive(path: Path, label: str) -> dict[str, Any]:
file_count = 0
expanded = 0
names: set[str] = set()
members: dict[str, dict[str, Any]] = {}
source_manifest_bytes: bytes | None = None
def accept_name(name: str) -> str:
safe = _safe_relative(name.rstrip("/"), f"{label} member")
@@ -249,7 +423,7 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]:
if zipfile.is_zipfile(path):
with zipfile.ZipFile(path) as archive:
for member in archive.infolist():
accept_name(member.filename)
safe_name = accept_name(member.filename)
mode = (member.external_attr >> 16) & 0xFFFF
if mode and stat.S_ISLNK(mode):
raise MaterialsError(f"symlink in {label}: {member.filename}")
@@ -263,10 +437,24 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]:
raise MaterialsError(f"excessive compression ratio in {label}: {member.filename}")
file_count += 1
expanded += member.file_size
digest = hashlib.sha256()
chunks: list[bytes] = []
with archive.open(member) as handle:
while chunk := handle.read(1024 * 1024):
digest.update(chunk)
if safe_name == SOURCE_MANIFEST_NAME:
chunks.append(chunk)
members[safe_name] = {
"path": safe_name,
"sha256": digest.hexdigest(),
"size_bytes": member.file_size,
}
if safe_name == SOURCE_MANIFEST_NAME:
source_manifest_bytes = b"".join(chunks)
else:
with tarfile.open(path, "r:*") as archive:
for member in archive:
accept_name(member.name)
safe_name = accept_name(member.name)
if member.issym() or member.islnk():
raise MaterialsError(f"symlink/hardlink in {label}: {member.name}")
if member.isdir():
@@ -275,6 +463,22 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]:
raise MaterialsError(f"non-regular member in {label}: {member.name}")
file_count += 1
expanded += member.size
handle = archive.extractfile(member)
if handle is None:
raise MaterialsError(f"unreadable regular member in {label}: {member.name}")
digest = hashlib.sha256()
chunks = []
while chunk := handle.read(1024 * 1024):
digest.update(chunk)
if safe_name == SOURCE_MANIFEST_NAME:
chunks.append(chunk)
members[safe_name] = {
"path": safe_name,
"sha256": digest.hexdigest(),
"size_bytes": member.size,
}
if safe_name == SOURCE_MANIFEST_NAME:
source_manifest_bytes = b"".join(chunks)
except (tarfile.TarError, zipfile.BadZipFile, OSError) as exc:
raise MaterialsError(f"invalid source archive for {label}: {path}") from exc
if file_count < 1:
@@ -284,7 +488,12 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]:
compressed_size = path.stat().st_size
if compressed_size and expanded > compressed_size * MAX_COMPRESSION_RATIO:
raise MaterialsError(f"excessive compression ratio in {label}")
return {"file_count": file_count, "expanded_size_bytes": expanded}
return {
"file_count": file_count,
"expanded_size_bytes": expanded,
"members": members,
"source_manifest_bytes": source_manifest_bytes,
}
def _validate_sbom(payload: Mapping[str, Any], role: str, manifest_digest: str) -> dict[str, Any]:
@@ -295,19 +504,46 @@ def _validate_sbom(payload: Mapping[str, Any], role: str, manifest_digest: str)
raise MaterialsError(f"{role} SBOM package inventory is empty")
ids: list[str] = []
roots = 0
components: list[dict[str, Any]] = []
for index, raw in enumerate(packages):
package = _mapping(raw, f"{role} SBOM package {index}")
spdx_id = _text(package.get("SPDXID"), f"{role} SBOM package SPDXID")
ids.append(spdx_id)
if package.get("versionInfo") == manifest_digest:
roots += 1
continue
external_refs = package.get("externalRefs", [])
if external_refs is None:
external_refs = []
purls = sorted(
{
_text(reference.get("referenceLocator"), f"{role} SBOM purl")
for reference in (
_mapping(value, f"{role} SBOM external reference")
for value in _list(external_refs, f"{role} SBOM external references")
)
if reference.get("referenceType") == "purl"
}
)
components.append(
{
"name": _text(package.get("name"), f"{role} SBOM package name"),
"version": _text(package.get("versionInfo"), f"{role} SBOM package version"),
"spdx_id": spdx_id,
"purls": purls,
}
)
_unique(ids, f"{role} SBOM SPDXID")
if roots != 1:
raise MaterialsError(f"{role} SBOM must bind exactly one root package to {manifest_digest}")
creators = _mapping(payload.get("creationInfo"), f"{role} SBOM creationInfo").get("creators")
if not isinstance(creators, list) or not any("syft" in str(item).lower() for item in creators):
raise MaterialsError(f"{role} SBOM is not identified as a Syft document")
return {"package_count": len(packages), "root_manifest_digest": manifest_digest}
return {
"package_count": len(packages),
"root_manifest_digest": manifest_digest,
"components": components,
}
def _license_is_unresolved(item: Mapping[str, Any]) -> bool:
@@ -378,41 +614,158 @@ def _parse_trivy(
return occurrences, {"occurrence_count": len(occurrences), "image_id": config_digest}
def _offer_details(
def _validate_source_archive(
path: Path,
artifact_id: str,
candidate: Mapping[str, Any],
image_manifest_digests: Mapping[str, str],
profile: str,
as_of: datetime,
occurrences: Mapping[str, Mapping[str, Any]],
sbom_components: Mapping[str, Mapping[str, list[Mapping[str, Any]]]],
) -> dict[str, Any]:
if profile != "redistributable":
raise MaterialsError("written source offers are forbidden for internal-hosted profile")
offer = _load_json(path, f"source offer {artifact_id}")
if offer.get("schema_version") != OFFER_SCHEMA or offer.get("offer_id") != artifact_id:
raise MaterialsError(f"invalid source offer identity for {artifact_id}")
if offer.get("status") != "APPROVED" or offer.get("profile") != profile:
raise MaterialsError(f"source offer {artifact_id} is not approved for {profile}")
if offer.get("candidate_commit") != candidate["commit"]:
raise MaterialsError(f"source offer {artifact_id} candidate mismatch")
if offer.get("source_manifest_sha256") != candidate["source_manifest_sha256"]:
raise MaterialsError(f"source offer {artifact_id} source manifest mismatch")
if offer.get("image_manifest_digests") != image_manifest_digests:
raise MaterialsError(f"source offer {artifact_id} image manifest binding mismatch")
issuer = _mapping(offer.get("issuer"), f"source offer {artifact_id} issuer")
for key in ("legal_name", "approver_name", "approver_title", "contact"):
_text(issuer.get(key), f"source offer {artifact_id} issuer.{key}")
_text(offer.get("authorization_id"), f"source offer {artifact_id} authorization_id")
_text(offer.get("approval_provenance_sha256"), f"source offer {artifact_id} approval_provenance_sha256")
if not HEX64.fullmatch(str(offer["approval_provenance_sha256"])):
raise MaterialsError(f"source offer {artifact_id} has invalid approval provenance hash")
valid_from = _parse_time(offer.get("valid_from"), f"source offer {artifact_id} valid_from")
valid_until = _parse_time(offer.get("valid_until"), f"source offer {artifact_id} valid_until")
if not valid_from <= as_of < valid_until:
raise MaterialsError(f"source offer {artifact_id} is not active at manifest as_of")
covers = [_text(item, f"source offer {artifact_id} covers item") for item in _list(offer.get("covers"), "offer covers")]
_unique(covers, f"source offer {artifact_id} covers")
return {"valid_from": valid_from.isoformat(), "valid_until": valid_until.isoformat(), "covers": covers}
inspected = _inspect_archive(path, f"source material {artifact_id}")
raw_manifest = inspected.pop("source_manifest_bytes")
members = inspected.pop("members")
if raw_manifest is None:
raise MaterialsError(
f"source material {artifact_id} lacks root {SOURCE_MANIFEST_NAME}"
)
try:
manifest_text = raw_manifest.decode("utf-8")
except UnicodeDecodeError as exc:
raise MaterialsError(f"source material {artifact_id} has invalid embedded manifest") from exc
manifest = _loads_json_strict(
manifest_text, f"source material {artifact_id} embedded manifest"
)
expected_keys = {
"schema_version",
"source_id",
"candidate_commit",
"source_manifest_sha256",
"package_name",
"package_version",
"binary_coordinate",
"source_coordinate",
"image_manifest_digests",
"covers",
"files",
}
if set(manifest) != expected_keys:
raise MaterialsError(
f"source material {artifact_id} embedded manifest keys must be exactly {sorted(expected_keys)}"
)
if manifest.get("schema_version") != SOURCE_ARCHIVE_SCHEMA or manifest.get("source_id") != artifact_id:
raise MaterialsError(f"source material {artifact_id} embedded identity mismatch")
if manifest.get("candidate_commit") != candidate["commit"]:
raise MaterialsError(f"source material {artifact_id} candidate mismatch")
if manifest.get("source_manifest_sha256") != candidate["source_manifest_sha256"]:
raise MaterialsError(f"source material {artifact_id} source manifest mismatch")
package_name = _text(manifest.get("package_name"), f"source material {artifact_id} package_name")
package_version = _text(manifest.get("package_version"), f"source material {artifact_id} package_version")
binary_coordinate = _text(
manifest.get("binary_coordinate"), f"source material {artifact_id} binary_coordinate"
)
source_coordinate = _text(
manifest.get("source_coordinate"), f"source material {artifact_id} source_coordinate"
)
if source_coordinate.upper() in {"UNKNOWN", "NOASSERTION"} or not re.match(
r"^(?:pkg:|https://|git\+https://)", source_coordinate
):
raise MaterialsError(f"source material {artifact_id} has an invalid source coordinate")
covers = [
_text(item, f"source material {artifact_id} covers item")
for item in _list(manifest.get("covers"), f"source material {artifact_id} covers")
]
if not covers:
raise MaterialsError(f"source material {artifact_id} covers no occurrences")
_unique(covers, f"source material {artifact_id} covers")
covered_roles: set[str] = set()
for occurrence_id in covers:
occurrence = occurrences.get(occurrence_id)
if occurrence is None:
raise MaterialsError(f"source material {artifact_id} covers unknown occurrence {occurrence_id}")
if str(occurrence["category"]).strip().lower() != "restricted":
raise MaterialsError(f"source material {artifact_id} covers non-restricted occurrence {occurrence_id}")
if occurrence["package_name"] != package_name:
raise MaterialsError(
f"source material {artifact_id} package/Trivy mismatch for {occurrence_id}"
)
role = str(occurrence["role"])
covered_roles.add(role)
candidates = sbom_components.get(role, {}).get(package_name, [])
identities = {
(str(item["version"]), purl)
for item in candidates
for purl in item.get("purls", [])
if item.get("version") not in {"UNKNOWN", "NOASSERTION", ""}
}
if identities != {(package_version, binary_coordinate)}:
raise MaterialsError(
f"source material {artifact_id} package version/coordinate does not exactly bind "
f"{role}/{package_name}: observed={sorted(identities)}"
)
expected_image_binding = {
role: image_manifest_digests[role] for role in sorted(covered_roles)
}
if manifest.get("image_manifest_digests") != expected_image_binding:
raise MaterialsError(f"source material {artifact_id} image binding mismatch")
declared_files: dict[str, dict[str, Any]] = {}
has_substantive_source = False
for index, raw in enumerate(_list(manifest.get("files"), f"source material {artifact_id} files")):
record = _mapping(raw, f"source material {artifact_id} file {index}")
if set(record) != {"path", "sha256", "size_bytes", "kind"}:
raise MaterialsError(f"source material {artifact_id} file records require path/hash/size/kind")
relative = _safe_relative(record.get("path"), f"source material {artifact_id} file path")
if relative == SOURCE_MANIFEST_NAME or relative in declared_files:
raise MaterialsError(f"duplicate/reserved source file in {artifact_id}: {relative}")
digest = _text(record.get("sha256"), f"source material {artifact_id} file hash")
size = record.get("size_bytes")
kind = _text(record.get("kind"), f"source material {artifact_id} file kind")
if not HEX64.fullmatch(digest) or not isinstance(size, int) or isinstance(size, bool) or size < 1:
raise MaterialsError(f"invalid embedded source file record in {artifact_id}: {relative}")
if kind not in {"source", "build", "patch", "license", "metadata", "source-archive"}:
raise MaterialsError(f"invalid embedded source file kind in {artifact_id}: {kind}")
basename = PurePosixPath(relative).name.lower()
documentation_only = basename.startswith(
("readme", "license", "copying", "notice", "authors", "changelog")
)
if kind == "source" and not documentation_only:
has_substantive_source = True
declared_files[relative] = {
"path": relative,
"sha256": digest,
"size_bytes": size,
"kind": kind,
}
actual_files = {key: value for key, value in members.items() if key != SOURCE_MANIFEST_NAME}
if set(declared_files) != set(actual_files):
raise MaterialsError(
f"source material {artifact_id} embedded file set mismatch: "
f"missing={sorted(set(declared_files) - set(actual_files))}, "
f"extra={sorted(set(actual_files) - set(declared_files))}"
)
for relative, declared in declared_files.items():
actual = actual_files[relative]
if (declared["sha256"], declared["size_bytes"]) != (
actual["sha256"],
actual["size_bytes"],
):
raise MaterialsError(f"source material {artifact_id} embedded file hash/size mismatch: {relative}")
if not has_substantive_source:
raise MaterialsError(
f"source material {artifact_id} contains documentation only, not corresponding source"
)
return {
**inspected,
"package_name": package_name,
"package_version": package_version,
"binary_coordinate": binary_coordinate,
"source_coordinate": source_coordinate,
"image_manifest_digests": expected_image_binding,
"covers": sorted(covers),
"files": [declared_files[key] for key in sorted(declared_files)],
"embedded_manifest_sha256": _sha256_bytes(raw_manifest),
}
def _compile(input_manifest: Path) -> dict[str, bytes]:
@@ -423,7 +776,16 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
raw_manifest = _load_json(manifest_path, "input manifest")
if raw_manifest.get("schema_version") != INPUT_SCHEMA:
raise MaterialsError(f"input manifest schema must be {INPUT_SCHEMA}")
expected_top = {"schema_version", "profile", "as_of", "candidate", "license_texts", "source_materials", "coverage"}
expected_top = {
"schema_version",
"profile",
"as_of",
"candidate",
"spdx_catalog",
"license_texts",
"source_materials",
"coverage",
}
if set(raw_manifest) != expected_top:
raise MaterialsError(f"input manifest keys must be exactly {sorted(expected_top)}")
profile = _text(raw_manifest.get("profile"), "profile")
@@ -452,6 +814,11 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
all_occurrences: list[dict[str, Any]] = []
image_summaries: list[dict[str, Any]] = []
input_records: list[dict[str, Any]] = []
spdx_license_ids, spdx_exception_ids, spdx_records = _load_spdx_catalog(
root, raw_manifest.get("spdx_catalog")
)
input_records.extend(spdx_records)
sbom_components_by_role: dict[str, dict[str, list[Mapping[str, Any]]]] = {}
roles: list[str] = []
manifest_digests: list[str] = []
config_digests: list[str] = []
@@ -471,6 +838,10 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
sbom_record, sbom_path = _file_record(root, image.get("sbom"), f"{role} SBOM")
trivy_record, trivy_path = _file_record(root, image.get("trivy_license"), f"{role} Trivy report")
sbom_summary = _validate_sbom(_load_json(sbom_path, f"{role} SBOM"), role, manifest_digest)
components_by_name: dict[str, list[Mapping[str, Any]]] = {}
for component in sbom_summary["components"]:
components_by_name.setdefault(component["name"], []).append(component)
sbom_components_by_role[role] = components_by_name
occurrences, trivy_summary = _parse_trivy(
_load_json(trivy_path, f"{role} Trivy report"), role, config_digest, trivy_record["sha256"]
)
@@ -505,18 +876,54 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
occurrence_by_id = {item["occurrence_id"]: item for item in all_occurrences}
license_defs: dict[str, dict[str, Any]] = {}
license_ast_by_id: dict[str, tuple[Any, ...]] = {}
license_aliases: dict[str, tuple[Any, ...]] = {}
license_payloads: dict[str, bytes] = {}
license_paths: list[str] = []
for index, raw_license in enumerate(_list(raw_manifest.get("license_texts"), "license_texts")):
record = _mapping(raw_license, f"license_texts[{index}]")
if set(record) != {"id", "kind", "file"}:
raise MaterialsError("license text records require exactly id, kind and file")
if set(record) != {"id", "kind", "spdx_expression", "aliases", "file"}:
raise MaterialsError(
"license text records require exactly id, kind, spdx_expression, aliases and file"
)
artifact_id = _text(record.get("id"), f"license_texts[{index}].id")
if not SAFE_ID.fullmatch(artifact_id) or artifact_id in license_defs:
raise MaterialsError(f"invalid or duplicate license text id: {artifact_id}")
kind = _text(record.get("kind"), f"license_texts[{index}].kind")
if kind not in {"license", "copyright", "notice"}:
raise MaterialsError(f"invalid license text kind: {kind}")
expression_ast = _parse_spdx_expression(
record.get("spdx_expression"),
f"license text {artifact_id} SPDX expression",
spdx_license_ids,
spdx_exception_ids,
)
aliases = [
_text(value, f"license text {artifact_id} alias")
for value in _list(record.get("aliases"), f"license text {artifact_id} aliases")
]
_unique(aliases, f"license text {artifact_id} aliases")
for alias in aliases:
lowered = alias.lower()
if any(marker in lowered for marker in ("licenseref-", "unknown", "unresolved", "noassertion")):
raise MaterialsError(f"unsafe SPDX alias for license text {artifact_id}: {alias}")
previous = license_aliases.get(alias)
if previous is not None and previous != expression_ast:
raise MaterialsError(f"SPDX alias maps to conflicting expressions: {alias}")
try:
parsed_alias = _parse_spdx_expression(
alias,
f"license text {artifact_id} alias",
spdx_license_ids,
spdx_exception_ids,
)
except MaterialsError:
parsed_alias = None
if parsed_alias is not None and parsed_alias != expression_ast:
raise MaterialsError(
f"canonical SPDX alias has conflicting meaning for license text {artifact_id}: {alias}"
)
license_aliases[alias] = expression_ast
file_record, path = _file_record(root, record.get("file"), f"license text {artifact_id}")
if file_record["size_bytes"] > MAX_LICENSE_TEXT:
raise MaterialsError(f"license text too large: {artifact_id}")
@@ -532,9 +939,12 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
license_defs[artifact_id] = {
"id": artifact_id,
"kind": kind,
"spdx_expression": _spdx_canonical(expression_ast),
"aliases": sorted(aliases),
"input": file_record,
"output_path": output_path,
}
license_ast_by_id[artifact_id] = expression_ast
license_payloads[output_path] = payload
if not license_defs:
raise MaterialsError("at least one license text is required")
@@ -556,14 +966,20 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
if kind not in {"source-archive", "written-source-offer"}:
raise MaterialsError(f"invalid source material kind: {kind}")
file_record, path = _file_record(root, record.get("file"), f"source material {artifact_id}")
if kind == "source-archive":
details = _validate_archive(path, f"source material {artifact_id}")
output_path = f"corresponding-source/{artifact_id}--{file_record['sha256'][:16]}.source"
else:
details = _offer_details(
path, artifact_id, candidate, image_manifest_digest_map, profile, as_of
if kind == "written-source-offer":
raise MaterialsError(
"written source offers are disabled until a pinned Ed25519 root/keyring and "
"authorized-signature policy are implemented; provide a real source archive"
)
output_path = f"source-offers/{artifact_id}--{file_record['sha256'][:16]}.json"
details = _validate_source_archive(
path,
artifact_id,
candidate,
image_manifest_digest_map,
occurrence_by_id,
sbom_components_by_role,
)
output_path = f"corresponding-source/{artifact_id}--{file_record['sha256'][:16]}.source"
source_paths.append(file_record["path"])
source_defs[artifact_id] = {
"id": artifact_id,
@@ -583,8 +999,16 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
used_source_ids: set[str] = set()
for index, raw_coverage in enumerate(coverage_list):
coverage = _mapping(raw_coverage, f"coverage[{index}]")
if set(coverage) != {"occurrence_id", "license_text_ids", "source_material_ids"}:
raise MaterialsError("coverage records require exactly occurrence_id, license_text_ids and source_material_ids")
if set(coverage) != {
"occurrence_id",
"spdx_expression",
"license_text_ids",
"source_material_ids",
}:
raise MaterialsError(
"coverage records require exactly occurrence_id, spdx_expression, "
"license_text_ids and source_material_ids"
)
occurrence_id = _text(coverage.get("occurrence_id"), f"coverage[{index}].occurrence_id")
if occurrence_id in coverage_by_id:
raise MaterialsError(f"duplicate coverage record: {occurrence_id}")
@@ -604,18 +1028,66 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
occurrence = occurrence_by_id.get(occurrence_id)
if occurrence is None:
raise MaterialsError(f"extra coverage record: {occurrence_id}")
raw_detected_expression = str(occurrence["license_name"]).strip()
try:
detected_ast = _parse_spdx_expression(
raw_detected_expression,
f"Trivy occurrence {occurrence_id} license_name",
spdx_license_ids,
spdx_exception_ids,
)
except MaterialsError as exc:
detected_ast = license_aliases.get(raw_detected_expression)
if detected_ast is None:
raise MaterialsError(
f"Trivy occurrence {occurrence_id} has no exact canonical SPDX or declared alias mapping: "
f"{raw_detected_expression}"
) from exc
coverage_ast = _parse_spdx_expression(
coverage.get("spdx_expression"),
f"coverage {occurrence_id} SPDX expression",
spdx_license_ids,
spdx_exception_ids,
)
if coverage_ast != detected_ast:
raise MaterialsError(
f"coverage {occurrence_id} SPDX expression does not exactly match Trivy AND/OR semantics"
)
selected_license_asts = [
license_ast_by_id[item]
for item in license_ids
if license_defs[item]["kind"] == "license"
]
required_terms = _spdx_terms(detected_ast)
selected_terms: set[str] = set()
for selected_ast in selected_license_asts:
terms = _spdx_terms(selected_ast)
if selected_ast != detected_ast and selected_ast[0] not in {"id", "with"}:
raise MaterialsError(
f"coverage {occurrence_id} uses a license text with different SPDX AND/OR semantics"
)
if not terms.issubset(required_terms):
raise MaterialsError(
f"coverage {occurrence_id} uses a license text for a different SPDX license or exception"
)
selected_terms.update(terms)
if selected_terms != required_terms:
raise MaterialsError(
f"coverage {occurrence_id} license texts do not exactly cover its SPDX license terms"
)
if occurrence["category"].strip().lower() == "restricted" and not source_ids:
raise MaterialsError(f"restricted occurrence lacks corresponding source/source offer: {occurrence_id}")
if not any(license_defs[item]["kind"] in {"license", "copyright"} for item in license_ids):
raise MaterialsError(f"coverage {occurrence_id} lacks a license or copyright text")
for source_id in source_ids:
source = source_defs[source_id]
if source["kind"] == "written-source-offer" and occurrence_id not in source["details"]["covers"]:
raise MaterialsError(f"source offer {source_id} does not cover {occurrence_id}")
if occurrence_id not in source["details"]["covers"]:
raise MaterialsError(f"source archive {source_id} does not cover {occurrence_id}")
used_license_ids.update(license_ids)
used_source_ids.update(source_ids)
coverage_by_id[occurrence_id] = {
"occurrence_id": occurrence_id,
"spdx_expression": _spdx_canonical(detected_ast),
"license_text_ids": sorted(license_ids),
"source_material_ids": sorted(source_ids),
}
@@ -629,8 +1101,6 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
if used_source_ids != set(source_defs):
raise MaterialsError(f"unused/extra source artifacts: {sorted(set(source_defs) - used_source_ids)}")
for source_id, source in source_defs.items():
if source["kind"] != "written-source-offer":
continue
actually_covered = {
occurrence_id
for occurrence_id, coverage in coverage_by_id.items()
@@ -639,7 +1109,7 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
declared = set(source["details"]["covers"])
if declared != actually_covered:
raise MaterialsError(
f"source offer {source_id} coverage mismatch: "
f"source archive {source_id} coverage mismatch: "
f"missing={sorted(actually_covered - declared)}, extra={sorted(declared - actually_covered)}"
)
@@ -690,6 +1160,7 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
f"- Image role: `{occurrence['role']}`",
f"- Component: `{package}`",
f"- Detected license: `{occurrence['license_name']}`",
f"- Canonical SPDX expression: `{coverage['spdx_expression']}`",
f"- Category: `{occurrence['category']}`",
f"- License/copyright text IDs: `{', '.join(coverage['license_text_ids'])}`",
f"- Corresponding-source/source-offer IDs: `{', '.join(coverage['source_material_ids']) or 'not-required-by-technical-policy'}`",
@@ -742,7 +1213,7 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
"profile": profile,
"as_of": as_of_text,
"candidate": {**candidate, "images": sorted(image_summaries, key=lambda item: item["role"])},
"inputs": sorted(input_records, key=lambda item: (item["role"], item["kind"])),
"inputs": sorted(input_records, key=lambda item: (item.get("role", ""), item["kind"])),
"input_manifest": {
"sha256": input_manifest_hash,
"size_bytes": input_manifest_size,
@@ -750,6 +1221,10 @@ def _compile(input_manifest: Path) -> dict[str, bytes]:
"summary": matrix["summary"],
"output_files": output_records,
"result": "PASS_TECHNICAL_MATERIALS_COMPLETE",
"manifest_payload_hash_contract": (
"SHA-256 over UTF-8 canonical JSON with sorted keys, compact separators and trailing LF; "
"manifest_payload_sha256 excluded"
),
"distribution_clearance_claimed": False,
"claims_not_made": [
"legal opinion",
yuqei-infra/tests/test_third_party_materials.py
+196 -44
View File
@@ -46,9 +46,24 @@ def _record(path: Path, root: Path) -> dict[str, object]:
}
def _source_tar(path: Path, *, bad_kind: str | None = None) -> None:
def _source_tar(
path: Path,
*,
source_manifest: dict[str, object] | None = None,
files: dict[str, bytes] | None = None,
bad_kind: str | None = None,
) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with tarfile.open(path, "w:gz") as archive:
if source_manifest is not None:
manifest_data = tool._canonical_json(source_manifest)
manifest_member = tarfile.TarInfo(tool.SOURCE_MANIFEST_NAME)
manifest_member.size = len(manifest_data)
archive.addfile(manifest_member, io.BytesIO(manifest_data))
for name, data in sorted((files or {}).items()):
member = tarfile.TarInfo(name)
member.size = len(data)
archive.addfile(member, io.BytesIO(data))
if bad_kind == "traversal":
member = tarfile.TarInfo("../escape.txt")
data = b"escape"
@@ -69,6 +84,8 @@ def _source_tar(path: Path, *, bad_kind: str | None = None) -> None:
archive.addfile(member)
return
else:
if source_manifest is not None:
return
member = tarfile.TarInfo("source/README.txt")
data = b"corresponding source fixture\n"
member.size = len(data)
@@ -76,26 +93,50 @@ def _source_tar(path: Path, *, bad_kind: str | None = None) -> None:
def _sbom(role: str, manifest_digest: str) -> dict[str, object]:
packages = [
{
"name": f"{role}.oci.tar",
"SPDXID": f"SPDXRef-Root-{role}",
"versionInfo": manifest_digest,
"licenseDeclared": "NOASSERTION",
},
{
"name": f"fixture-{role}",
"SPDXID": f"SPDXRef-Package-{role}",
"versionInfo": "1.0.0",
"licenseDeclared": "MIT",
"externalRefs": [
{
"referenceCategory": "PACKAGE-MANAGER",
"referenceType": "purl",
"referenceLocator": f"pkg:generic/fixture-{role}@1.0.0",
}
],
},
]
if role == "contract-api":
packages.append(
{
"name": "fixture-copyleft",
"SPDXID": "SPDXRef-Package-fixture-copyleft",
"versionInfo": "1.0.0",
"licenseDeclared": "GPL-2.0-only",
"externalRefs": [
{
"referenceCategory": "PACKAGE-MANAGER",
"referenceType": "purl",
"referenceLocator": "pkg:generic/fixture-copyleft@1.0.0",
}
],
}
)
return {
"spdxVersion": "SPDX-2.3",
"SPDXID": "SPDXRef-DOCUMENT",
"name": f"{role}.oci.tar",
"documentNamespace": f"https://example.invalid/sbom/{role}",
"creationInfo": {"creators": ["Tool: syft-1.52.0"], "created": "2026-09-27T00:00:00Z"},
"packages": [
{
"name": f"{role}.oci.tar",
"SPDXID": f"SPDXRef-Root-{role}",
"versionInfo": manifest_digest,
"licenseDeclared": "NOASSERTION",
},
{
"name": f"fixture-{role}",
"SPDXID": f"SPDXRef-Package-{role}",
"versionInfo": "1.0.0",
"licenseDeclared": "MIT",
},
],
"packages": packages,
}
@@ -144,8 +185,24 @@ def _fixture(tmp_path: Path, *, profile: str = "redistributable") -> tuple[Path,
mit.parent.mkdir()
mit.write_text("MIT fixture license text\n", encoding="utf-8")
gpl.write_text("GPL-2.0-only fixture license text\n", encoding="utf-8")
source = root / "sources" / "fixture-copyleft.tar.gz"
_source_tar(source)
spdx_licenses = root / "spdx" / "licenses.json"
spdx_exceptions = root / "spdx" / "exceptions.json"
_write_json(
spdx_licenses,
{
"licenseListVersion": "3.29.0",
"licenses": [
{"licenseId": "Apache-2.0", "isDeprecatedLicenseId": False},
{"licenseId": "GPL-2.0-only", "isDeprecatedLicenseId": False},
{"licenseId": "MIT", "isDeprecatedLicenseId": False},
],
},
)
_write_json(
spdx_exceptions,
{"licenseListVersion": "3.29.0", "exceptions": []},
)
images: list[dict[str, object]] = []
occurrences: list[dict[str, object]] = []
@@ -176,10 +233,35 @@ def _fixture(tmp_path: Path, *, profile: str = "redistributable") -> tuple[Path,
coverage.append(
{
"occurrence_id": occurrence["occurrence_id"],
"spdx_expression": occurrence["license_name"],
"license_text_ids": ["GPL-2.0-only" if restricted else "MIT"],
"source_material_ids": ["fixture-copyleft-source"] if restricted else [],
}
)
restricted_occurrence = next(item for item in occurrences if item["category"] == "restricted")
contract_api_image = next(item for item in images if item["role"] == "contract-api")
source_bytes = b"def copyleft_fixture() -> str:\n return 'source'\n"
source_file_record = {
"path": "src/main.py",
"sha256": tool._sha256_bytes(source_bytes),
"size_bytes": len(source_bytes),
"kind": "source",
}
source_manifest = {
"schema_version": tool.SOURCE_ARCHIVE_SCHEMA,
"source_id": "fixture-copyleft-source",
"candidate_commit": "a" * 40,
"source_manifest_sha256": "b" * 64,
"package_name": "fixture-copyleft",
"package_version": "1.0.0",
"binary_coordinate": "pkg:generic/fixture-copyleft@1.0.0",
"source_coordinate": "pkg:generic/fixture-copyleft-source@1.0.0",
"image_manifest_digests": {"contract-api": contract_api_image["manifest_digest"]},
"covers": [restricted_occurrence["occurrence_id"]],
"files": [source_file_record],
}
source = root / "sources" / "fixture-copyleft.tar.gz"
_source_tar(source, source_manifest=source_manifest, files={"src/main.py": source_bytes})
manifest: dict[str, object] = {
"schema_version": tool.INPUT_SCHEMA,
"profile": profile,
@@ -190,9 +272,25 @@ def _fixture(tmp_path: Path, *, profile: str = "redistributable") -> tuple[Path,
"source_manifest_sha256": "b" * 64,
"images": images,
},
"spdx_catalog": {
"licenses": _record(spdx_licenses, root),
"exceptions": _record(spdx_exceptions, root),
},
"license_texts": [
{"id": "MIT", "kind": "license", "file": _record(mit, root)},
{"id": "GPL-2.0-only", "kind": "license", "file": _record(gpl, root)},
{
"id": "MIT",
"kind": "license",
"spdx_expression": "MIT",
"aliases": [],
"file": _record(mit, root),
},
{
"id": "GPL-2.0-only",
"kind": "license",
"spdx_expression": "GPL-2.0-only",
"aliases": [],
"file": _record(gpl, root),
},
],
"source_materials": [
{
@@ -331,7 +429,12 @@ def test_coverage_must_be_an_exact_nonduplicated_set(tmp_path: Path, mode: str)
expected = "duplicate coverage record"
else:
coverage.append(
{"occurrence_id": "occ-" + "f" * 64, "license_text_ids": ["MIT"], "source_material_ids": []}
{
"occurrence_id": "occ-" + "f" * 64,
"spdx_expression": "MIT",
"license_text_ids": ["MIT"],
"source_material_ids": [],
}
)
expected = "extra coverage record"
_rewrite_manifest(manifest_path, manifest)
@@ -424,32 +527,85 @@ def test_restricted_occurrence_requires_source_and_unused_artifacts_fail(tmp_pat
tool.build(manifest_path, tmp_path / "materials-extra")
def test_valid_written_offer_is_bound_and_internal_profile_rejects_it(tmp_path: Path) -> None:
def test_mit_text_cannot_masquerade_as_gpl_coverage(tmp_path: Path) -> None:
manifest_path, manifest = _fixture(tmp_path)
restricted = next(item for item in manifest["coverage"] if item["source_material_ids"])
restricted["license_text_ids"] = ["MIT"]
_rewrite_manifest(manifest_path, manifest)
with pytest.raises(tool.MaterialsError, match="different SPDX license|do not exactly cover"):
tool.build(manifest_path, tmp_path / "materials")
def test_license_text_and_or_semantics_cannot_masquerade(tmp_path: Path) -> None:
manifest_path, manifest = _fixture(tmp_path)
manifest["license_texts"][0]["spdx_expression"] = "MIT OR Apache-2.0"
_rewrite_manifest(manifest_path, manifest)
with pytest.raises(tool.MaterialsError, match="different SPDX AND/OR semantics"):
tool.build(manifest_path, tmp_path / "materials")
def test_readme_only_archive_cannot_masquerade_as_corresponding_source(tmp_path: Path) -> None:
manifest_path, manifest = _fixture(tmp_path)
root = manifest_path.parent
source_record = manifest["source_materials"][0]["file"]
source_path = root / str(source_record["path"])
with tarfile.open(source_path, "r:gz") as archive:
member = archive.getmember(tool.SOURCE_MANIFEST_NAME)
extracted = archive.extractfile(member)
assert extracted is not None
source_manifest = json.loads(extracted.read().decode("utf-8"))
readme = b"This is not corresponding source.\n"
source_manifest["files"] = [
{
"path": "README.md",
"sha256": tool._sha256_bytes(readme),
"size_bytes": len(readme),
"kind": "source",
}
]
_source_tar(source_path, source_manifest=source_manifest, files={"README.md": readme})
_update_file_record(root, source_record)
_rewrite_manifest(manifest_path, manifest)
with pytest.raises(tool.MaterialsError, match="documentation only"):
tool.build(manifest_path, tmp_path / "materials")
def test_source_archive_package_version_must_match_sbom_identity(tmp_path: Path) -> None:
manifest_path, manifest = _fixture(tmp_path)
root = manifest_path.parent
source_record = manifest["source_materials"][0]["file"]
source_path = root / str(source_record["path"])
with tarfile.open(source_path, "r:gz") as archive:
embedded = archive.extractfile(archive.getmember(tool.SOURCE_MANIFEST_NAME))
source_file = archive.extractfile(archive.getmember("src/main.py"))
assert embedded is not None and source_file is not None
source_manifest = json.loads(embedded.read().decode("utf-8"))
source_bytes = source_file.read()
source_manifest["package_version"] = "9.9.9"
_source_tar(source_path, source_manifest=source_manifest, files={"src/main.py": source_bytes})
_update_file_record(root, source_record)
_rewrite_manifest(manifest_path, manifest)
with pytest.raises(tool.MaterialsError, match="version/coordinate does not exactly bind"):
tool.build(manifest_path, tmp_path / "materials")
def test_self_asserted_written_offer_fails_without_pinned_ed25519_trust(tmp_path: Path) -> None:
manifest_path, manifest = _fixture(tmp_path)
root = manifest_path.parent
restricted = next(item for item in manifest["coverage"] if item["source_material_ids"])
offer_id = "approved-source-offer"
offer_id = "self-signed-source-offer"
offer = {
"schema_version": tool.OFFER_SCHEMA,
"schema_version": "yuqei.written-source-offer.v1",
"offer_id": offer_id,
"status": "APPROVED",
"profile": "redistributable",
"candidate_commit": manifest["candidate"]["commit"],
"source_manifest_sha256": manifest["candidate"]["source_manifest_sha256"],
"image_manifest_digests": {
image["role"]: image["manifest_digest"] for image in manifest["candidate"]["images"]
},
"issuer": {
"legal_name": "Example Fixture Organization",
"approver_name": "Fixture Approver",
"approver_title": "Release Owner",
"contact": "source-offer@example.invalid",
},
"authorization_id": "FIXTURE-OFFER-001",
"approval_provenance_sha256": "c" * 64,
"valid_from": "2026-09-01T00:00:00Z",
"valid_until": "2027-09-01T00:00:00Z",
"covers": [restricted["occurrence_id"]],
"public_key": "attacker-controlled-key",
"signature": "attacker-controlled-signature",
}
offer_path = root / "offers" / "offer.json"
_write_json(offer_path, offer)
@@ -459,12 +615,8 @@ def test_valid_written_offer_is_bound_and_internal_profile_rejects_it(tmp_path:
restricted["source_material_ids"] = [offer_id]
_rewrite_manifest(manifest_path, manifest)
tool.build(manifest_path, tmp_path / "offer-materials")
manifest["profile"] = "internal-hosted"
_rewrite_manifest(manifest_path, manifest)
with pytest.raises(tool.MaterialsError, match="forbidden for internal-hosted"):
tool.build(manifest_path, tmp_path / "internal-materials")
with pytest.raises(tool.MaterialsError, match="pinned Ed25519 root/keyring"):
tool.build(manifest_path, tmp_path / "offer-materials")
def test_verifier_rejects_extra_or_modified_output(tmp_path: Path) -> None: