← back to Dw Five Field Step0

sku_guard.py

64 lines

"""Fail-closed DW SKU identity guard anchored to fetched live variants."""

import re

ASCII_SKU = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$", re.ASCII)
SAMPLE_SUFFIX = re.compile(r"-sample$", re.IGNORECASE | re.ASCII)


def _strict_sku(value):
    if not isinstance(value, str) or value != value.strip() or not ASCII_SKU.fullmatch(value):
        return None
    if value.lower() in {"none", "null", "null-sample"}:
        return None
    return value


def validate_dw_sku(candidate, lane, live_variants):
    """Require a lane-shaped target derived from an actual fetched live variant SKU."""
    target = _strict_sku(candidate)
    if not target:
        return {"ok": False, "reason": "blank-candidate", "basis": None}
    if not isinstance(live_variants, list):
        return {"ok": False, "reason": "invalid-live-variants", "basis": None}

    if lane == "add-sample":
        if not target.endswith("-Sample"):
            return {"ok": False, "reason": "invalid-sample-target", "basis": None}
        target_base = target[:-7]
        if target_base != target_base.upper():
            return {"ok": False, "reason": "noncanonical-sample-base", "basis": None}
        trusted = []
        for variant in live_variants:
            if not isinstance(variant, dict) or str(variant.get("option1") or "").strip().lower() == "sample":
                continue
            live_sku = _strict_sku(variant.get("sku"))
            if live_sku and live_sku == live_sku.upper():
                trusted.append(live_sku)
        if len(trusted) != 1:
            return {"ok": False, "reason": "ambiguous-live-non-sample-identity", "basis": None}
        if target_base == trusted[0]:
            return {"ok": True, "reason": "derived-from-live-non-sample", "basis": "live-variant"}
        return {"ok": False, "reason": "sample-base-not-live", "basis": None}

    if lane == "build-roll":
        if target != target.upper():
            return {"ok": False, "reason": "noncanonical-roll-base", "basis": None}
        trusted = []
        for variant in live_variants:
            if not isinstance(variant, dict):
                continue
            live_sku = _strict_sku(variant.get("sku"))
            is_sample = str(variant.get("option1") or "").strip().lower() == "sample"
            if live_sku and is_sample and SAMPLE_SUFFIX.search(live_sku):
                live_base = SAMPLE_SUFFIX.sub("", live_sku)
                if live_base == live_base.upper():
                    trusted.append(live_base)
        if len(trusted) != 1:
            return {"ok": False, "reason": "ambiguous-live-sample-identity", "basis": None}
        if target == trusted[0]:
            return {"ok": True, "reason": "recovered-from-live-sample", "basis": "live-sample-variant"}
        return {"ok": False, "reason": "roll-base-not-live-sample", "basis": None}

    return {"ok": False, "reason": "unknown-lane", "basis": None}