CASAN/AINative_OKR_CASAN5/packages/casan-harness/scripts/bash/pii-mask.py

#!/usr/bin/env python3
"""CASAN H4 PII masker driven by packages/casan-harness/security/pii-rules.yaml.

Reads content on stdin, applies every `action: mask` rule from the rules
file, and writes the masked content to stdout. Type-specific replacement
tokens are preserved so downstream evidence stays stable
(***MASKED_EMAIL***, ***MASKED_PHONE***, ***MASKED_ID***).

This makes pii-rules.yaml the source of truth for PII masking instead of
dead config: editing/removing a rule changes runtime behavior.
"""
import re
import sys

REPLACEMENT_BY_TYPE = {
    "email": "***MASKED_EMAIL***",
    "phone": "***MASKED_PHONE***",
    "personal_id": "***MASKED_ID***",
    "address": "***MASKED_ADDRESS***",
}


def load_rules(path):
    rules, cur = [], {}
    with open(path, encoding="utf-8") as fh:
        for raw in fh:
            s = raw.strip()
            m = re.match(r"-\s*id:\s*(\S+)", s)
            if m:
                if cur:
                    rules.append(cur)
                cur = {"id": m.group(1)}
                continue
            m = re.match(r'type:\s*"?([^"\s]+)"?', s)
            if m:
                cur["type"] = m.group(1)
                continue
            m = re.match(r'regex:\s*"(.*)"\s*$', s)
            if m:
                # YAML double-quoted: collapse \\ -> \ to recover the real regex.
                cur["regex"] = m.group(1).replace("\\\\", "\\")
                continue
            m = re.match(r"action:\s*(\S+)", s)
            if m:
                cur["action"] = m.group(1)
                continue
    if cur:
        rules.append(cur)
    return rules


def main():
    data = sys.stdin.read()
    # SEC-08 (M-06): FAIL CLOSED. Previously a missing rules file, an unreadable
    # file, or a broken rule regex all emitted the RAW data — so a mask rule that
    # failed to load silently leaked the PII it was meant to hide. Now any such
    # condition emits NOTHING and exits non-zero: no unmasked content ever escapes.
    if len(sys.argv) < 2:
        sys.stderr.write("PII_MASK_FAIL no rules file provided (fail-closed)\n")
        return 1
    try:
        rules = load_rules(sys.argv[1])
    except OSError as exc:
        sys.stderr.write(f"PII_MASK_FAIL rules file unreadable (fail-closed): {exc}\n")
        return 1
    # Pre-compile every mask rule; a broken regex is fatal (that PII type would
    # otherwise pass through unmasked). Validate all BEFORE emitting anything.
    compiled = []
    for rule in rules:
        if rule.get("action") != "mask" or "regex" not in rule:
            continue
        token = REPLACEMENT_BY_TYPE.get(rule.get("type", ""), "***MASKED***")
        try:
            compiled.append((re.compile(rule["regex"]), token))
        except re.error as exc:
            sys.stderr.write(f"PII_MASK_FAIL bad regex in rule {rule.get('id')} (fail-closed): {exc}\n")
            return 1
    for rx, token in compiled:
        data = rx.sub(token, data)
    sys.stdout.write(data)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())