XFESystem

"""Collect locked C# dependency notices without modifying NuGet packages.

Use an already restored cache. --Fetch permits only the pinned, hashed official
GitHub license URLs in LicenseSources.json; otherwise collection is offline.
Existing output must be byte-identical, so reruns cannot silently replace notices.
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import json
import re
import urllib.request
import xml.etree.ElementTree as ET
import zipfile
from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]
OUTPUT = ROOT / "Tools/CSharpBoot/Licenses/Collected"
SOURCES = ROOT / "Tools/CSharpBoot/LicenseSources.json"
KERNEL_LOCK = ROOT / "Source/XFESystem.Kernel/Packages.lock.json"


def sha(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def save(relative: str, data: bytes) -> dict:
    target = (OUTPUT / relative).resolve()
    if not target.is_relative_to(OUTPUT.resolve()):
        raise ValueError("License output escaped its dedicated directory")
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.exists():
        if target.read_bytes() != data:
            raise ValueError(f"Existing notice differs: {target}")
    else:
        with target.open("xb") as stream:
            stream.write(data)
    return {"Path": relative, "Sha256": sha(data), "Bytes": len(data)}


def read_json(path: Path):
    return json.loads(path.read_text(encoding="utf-8-sig"))


def package_license_key(identifier: str) -> str:
    identifier = identifier.lower()
    if "cosmos." in identifier:
        return "Cosmos"
    if identifier.startswith("microsoft.codeanalysis.analyzers"):
        return "RoslynAnalyzers"
    if identifier.startswith("microsoft.codeanalysis"):
        return "Roslyn"
    return {
        "acornima": "Acornima", "anglesharp": "AngleSharp",
        "anglesharp.css": "AngleSharpCss", "htmlsanitizer": "HtmlSanitizer",
        "humanizer.core": "Humanizer", "jint": "Jint", "markdig": "Markdig",
        "mono.cecil": "MonoCecil",
    }.get(identifier, "DotNetRuntime" if identifier.startswith(("microsoft.", "runtime.", "system.")) else "")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--NuGetCache", type=Path, default=Path.home() / ".nuget/packages")
    parser.add_argument("--Fetch", action="store_true")
    args = parser.parse_args()
    specifications = read_json(SOURCES)
    official = {}
    files = []
    for entry in specifications["OfficialSources"]:
        relative = f"Upstream/{entry['Name']}/License.txt"
        target = OUTPUT / relative
        if target.exists():
            content = target.read_bytes()
        elif args.Fetch:
            url = entry["Url"]
            if not url.startswith("https://raw.githubusercontent.com/"):
                raise ValueError("Only fixed official GitHub raw sources may be fetched")
            request = urllib.request.Request(url, headers={"User-Agent": "XFESystem-LicenseCollector"})
            with urllib.request.urlopen(request, timeout=30) as response:
                if not response.url.startswith("https://raw.githubusercontent.com/"):
                    raise ValueError("Unexpected license download redirect")
                content = response.read(1_048_577)
        else:
            raise FileNotFoundError(f"Missing {relative}; use --Fetch for the pinned official source")
        if sha(content) != entry["Sha256"]:
            raise ValueError(f"Official license hash mismatch: {entry['Name']}")
        record = save(relative, content) | {"Source": entry["Url"], "Note": entry.get("Note", "")}
        files.append(record)
        official[entry["Name"]] = record["Path"]

    for entry in specifications["RepositoryFiles"]:
        source_root = args.NuGetCache.resolve() if "NuGetFile" in entry else ROOT
        source_path = (source_root / (entry["NuGetFile"] if "NuGetFile" in entry else entry["Source"])).resolve()
        if not source_path.is_relative_to(source_root):
            raise ValueError("Local license source escaped its repository or NuGet cache")
        source_description = "NuGetCache:" + entry["NuGetFile"] if "NuGetFile" in entry else entry["Source"]
        content = source_path.read_bytes()
        if sha(content) != entry["Sha256"]:
            raise ValueError(f"Local license changed: {source_description}")
        record = save(entry["Destination"], content) | {"Source": source_description}
        files.append(record)
        if "Key" in entry:
            official[entry["Key"]] = record["Path"]

    lock = read_json(KERNEL_LOCK)
    dependencies = {}
    for framework in lock["dependencies"].values():
        for identifier, dependency in framework.items():
            if dependency["type"] != "Project":
                old = dependencies.get(identifier)
                current = (dependency["resolved"], dependency["contentHash"])
                if old is not None and old != current:
                    raise ValueError(f"Conflicting locked package: {identifier}")
                dependencies[identifier] = current
    for item in specifications["ImplicitRuntimePackages"]:
        dependencies[item["Id"]] = (item["Version"], None)

    packages = []
    for identifier, (version, expected_hash) in sorted(dependencies.items()):
        if not re.fullmatch(r"[A-Za-z0-9_.-]+", identifier) or not re.fullmatch(r"[A-Za-z0-9_.+-]+", version):
            raise ValueError("Invalid locked package identifier or version")
        cache = args.NuGetCache / identifier.lower() / version
        archive = cache / f"{identifier.lower()}.{version}.nupkg"
        package_bytes = archive.read_bytes()
        zip_sha512 = base64.b64encode(hashlib.sha512(package_bytes).digest()).decode("ascii")
        sidecar = archive.with_suffix(".nupkg.sha512").read_text(encoding="utf-8-sig").strip()
        if sidecar != zip_sha512:
            raise ValueError(f"NuGet archive differs from its SHA512 sidecar: {identifier}")
        metadata = read_json(cache / ".nupkg.metadata")
        if expected_hash is not None and metadata["contentHash"] != expected_hash:
            raise ValueError(f"NuGet metadata differs from the locked content hash: {identifier}")
        prefix = f"Packages/{identifier}-{version}"
        notices = []
        with zipfile.ZipFile(archive) as package:
            nuspec_name = next(name for name in package.namelist() if name.lower().endswith(".nuspec"))
            nuspec = package.read(nuspec_name)
            xml = ET.fromstring(nuspec)
            tags = {element.tag.rsplit("}", 1)[-1]: element for element in xml.iter()}
            license_tag = tags.get("license")
            license_value = license_tag.text if license_tag is not None else None
            license_type = license_tag.get("type") if license_tag is not None else None
            record = save(f"{prefix}/Metadata.nuspec", nuspec) | {"Source": f"nuget:{identifier}/{version}/{nuspec_name}"}
            files.append(record)
            for name in package.namelist():
                basename = Path(name).name.lower()
                is_notice = any(term in basename for term in ("license", "notice", "copying"))
                is_declared = license_type == "file" and name.lower() == (license_value or "").lower()
                if name.endswith("/") or not (is_notice or is_declared):
                    continue
                if Path(basename).suffix not in ("", ".txt", ".md", ".rtf", ".html"):
                    continue
                safe_name = name.replace("/", "-").replace("\\", "-")
                safe_name = safe_name[0].upper() + safe_name[1:]
                record = save(f"{prefix}/{safe_name}", package.read(name)) | {"Source": f"nuget:{identifier}/{version}/{name}"}
                files.append(record)
                notices.append(record["Path"])
        key = package_license_key(identifier)
        if key:
            notices.append(official[key])
        if not notices:
            raise ValueError(f"No complete license text was collected for {identifier}")
        repository = tags.get("repository")
        packages.append({
            "Id": identifier, "Version": version,
            "Scope": "LockedKernelGraphIncludingBuildTools" if expected_hash else "ImplicitNativeAotRuntimePack",
            "NuGetSource": metadata.get("source"),
            "LockedContentHash": expected_hash,
            "NuGetContentHashFromMetadata": metadata["contentHash"],
            "ContentHashNote": "NuGet normalized package content hash; archive SHA256/SHA512 are recorded separately",
            "NupkgSha256": sha(package_bytes), "NupkgSha512": zip_sha512,
            "LicenseDeclaration": {"Type": license_type, "Value": license_value},
            "Repository": repository.attrib if repository is not None else None,
            "LicenseFiles": sorted(set(notices)),
        })
    manifest = {
        "Schema": "XFESystem.CSharpLicenses.V1",
        "KernelLockSha256": sha(KERNEL_LOCK.read_bytes()),
        "SourcesLockSha256": sha(SOURCES.read_bytes()),
        "PackageCount": len(packages), "Packages": packages,
        "Files": sorted(files, key=lambda entry: entry["Path"]),
        "ScopeNote": "Build tools are included conservatively; listing a dependency does not imply its code survives NativeAOT trimming or is executed in the guest",
    }
    save("Manifest.json", (json.dumps(manifest, ensure_ascii=False, indent=2) + "\n").encode("utf-8"))
    print(json.dumps({"Packages": len(packages), "Files": len(files), "Output": str(OUTPUT)}, ensure_ascii=False))


if __name__ == "__main__":
    main()