felhom.eu/documentation/audits/SPIKE-cc-creds-2026-10-17/probes/mask.py

#!/usr/bin/env python3
"""SPIKE probe — `sec run` shape: run a child with secrets in its ENVIRONMENT, mask its output.

    mask.py [--secret NAME=FILE]... [--stdin NAME] [--no-filter] -- CMD [ARGS...]

* each secret is read from FILE (trailing newline stripped) and put into the child's env as NAME;
* the child's stdout and stderr are filtered: every known encoding of each value becomes «NAME»;
* --stdin NAME feeds that value (plus "\n") to the child's stdin, then closes it;
* the child's exit code is passed through (128+signal when killed by a signal);
* --no-filter is the red-proof control: same env, same child, output untouched.

Forms masked, per value: raw; URL-encoded (quote, quote_plus); JSON-escaped; base64 and urlsafe base64
at all three byte alignments (the stable middle of the encoding, so a value embedded in a larger
base64 stream is still caught). A carry buffer of (longest form - 1) bytes is held back between reads,
so a value split across reads — even one byte per write — is still matched. The held-back tail is
released at EOF; a live-streaming command therefore lags by at most that many bytes.

NOT caught (by design of a byte filter): a value transformed in any other way (hex, rot13, reversed,
gzip, split by anything other than a line break). The design must not promise more.
"""
import base64
import json
import os
import re
import signal
import subprocess
import sys
import threading
import urllib.parse


def b64_stable(value, enc):
    """Stable substrings of enc(value) when value starts at byte offset 0, 1 or 2 of a larger stream."""
    out = []
    for k in range(3):
        e = enc(b"\0" * k + value)
        total = k + len(value)
        start = 4 if k else 0                 # first group mixes pad bytes with value bytes
        end = 4 * (total // 3)                # last partial group changes with the next byte
        s = e[start:end]
        if len(s) >= 8:
            out.append(s)
    return out


def forms(value):
    v = value.encode()
    f = {v,
         urllib.parse.quote(value, safe="").encode(),
         urllib.parse.quote_plus(value).encode(),
         json.dumps(value)[1:-1].encode()}
    f.update(b64_stable(v, base64.b64encode))
    f.update(b64_stable(v, base64.urlsafe_b64encode))
    return sorted((x for x in f if x), key=len, reverse=True)


def build_table(secrets):
    """(regex, token, form length) per form. A line break is allowed between any two bytes of a form:
    `base64` wraps at 76 columns, and the Q1 shape-13 red-proof showed 2 of 3 wrapped copies leaking
    through an exact-match filter. The carry buffer is sized for the longest form plus its breaks."""
    table = []
    for name, value in secrets:
        for fm in forms(value):
            rx = re.compile(b"(?:\r?\n)?".join(re.escape(bytes([c])) for c in fm))
            table.append((rx, ("«%s»" % name).encode(), len(fm)))
    table.sort(key=lambda t: t[2], reverse=True)
    return table


def pump(src, dst, table, filtering):
    keep = 3 * max((n for _, _, n in table), default=1)   # form + up to 2 bytes of line break per byte
    carry = b""
    while True:
        chunk = os.read(src, 65536)
        if not chunk:
            break
        if not filtering:
            dst.write(chunk); dst.flush(); continue
        buf = carry + chunk
        for rx, tok, _ in table:
            buf = rx.sub(tok, buf)
        if keep and len(buf) > keep:
            dst.write(buf[:-keep]); carry = buf[-keep:]
        elif keep:
            carry = buf
        else:
            dst.write(buf)
        dst.flush()
    if carry:
        for rx, tok, _ in table:
            carry = rx.sub(tok, carry)
        dst.write(carry)
    dst.flush()
    os.close(src)


def main(argv):
    secrets, stdin_name, filtering = [], None, True
    i = 0
    while i < len(argv) and argv[i] != "--":
        a = argv[i]
        if a == "--secret":
            name, path = argv[i + 1].split("=", 1)
            with open(path) as fh:
                secrets.append((name, fh.read().rstrip("\n")))
            i += 2
        elif a == "--stdin":
            stdin_name = argv[i + 1]; i += 2
        elif a == "--no-filter":
            filtering = False; i += 1
        else:
            sys.exit("mask.py: unknown option %r" % a)
    cmd = argv[i + 1:]
    if not cmd:
        sys.exit("mask.py: no command")
    env = dict(os.environ)
    env.update(dict(secrets))
    table = build_table(secrets)
    r_out, w_out = os.pipe()
    r_err, w_err = os.pipe()
    child = subprocess.Popen(cmd, env=env, stdout=w_out, stderr=w_err,
                             stdin=subprocess.PIPE if stdin_name else None)
    os.close(w_out); os.close(w_err)
    threads = [threading.Thread(target=pump, args=(r_out, sys.stdout.buffer, table, filtering)),
               threading.Thread(target=pump, args=(r_err, sys.stderr.buffer, table, filtering))]
    for t in threads:
        t.start()
    if stdin_name:
        child.stdin.write((dict(secrets)[stdin_name] + "\n").encode())
        child.stdin.close()
    rc = child.wait()
    for t in threads:
        t.join()
    return 128 - rc if rc < 0 else rc        # -SIGTERM -> 128+15


if __name__ == "__main__":
    signal.signal(signal.SIGPIPE, signal.SIG_DFL)
    sys.exit(main(sys.argv[1:]))