NUVLtech — Neutral Unified Verification Layer

Neutral Unified Verification Layer

The request path is not the authority.

"It should not require secrecy, and it should not be a problem if it falls into enemy hands."

Auguste Kerckhoffs, 1883

"The enemy knows the system," i.e., "one ought to design systems under the assumption that the enemy will immediately gain full familiarity with them".

Claude Shannon, 1949

Since Kerckhoffs, cryptography has had a clear discipline:

Publish the mechanism.

Protect the key.

Execution infrastructure still violates that discipline in practice.

Gateways, brokers, meshes, relays, queues, agents, and orchestration layers accumulate what they were never meant to hold: keys, sessions, policy, decision state, and practical authority. Each becomes a component that cannot fall into enemy hands without consequence.

The recurring failure pattern is the same. The attacker did not break the cryptography. The attacker captured something that had quietly become the authority.

NUVL applies the Kerckhoffs–Shannon discipline to the execution path itself.

The intermediary may be public, inspectable, exposed, copied, deployed on hardware, or placed in a hostile path without acquiring provider authorization authority.

Capture the intermediary and you still have not captured the provider boundary.

Requests may move.

Artifacts may move.

Authority stays with the provider.

Assume the intermediary is captured.

Not probed.

Not phished.

Captured.

Root access. Full memory. Complete source.

What does the attacker now hold?

The provider signing key — no. NUVL never has it.
Authorization policy — no. NUVL never evaluates it.
Session or authorization state — no. NUVL retains none.
Provider decisions — no. NUVL never learns the outcome.
The ability to initiate execution — no. That capability does not exist in the intermediary.

The attacker holds a relay.

The boundary is intact.

This is not hardening. Hardening lowers the probability of capture. NUVL removes the authorization payoff.

What activates before “no”?

In a conventional path, a request that was always going to be denied may still wake the gateway, activate the application, touch the database, hit context services, or trigger policy and enrichment logic before the denial happens.

The attacker pays nothing.

You pay for all of it — in compute, and in attack surface.

Provider-first ordering denies before downstream machinery exists to the request.

In instrumented comparisons, provider-first ordering showed zero database activation before denial, with roughly mid-20% latency reduction and low-30% CPU reduction in the tested path.

Exact values belong in dated lab reports, screenshots, dashboards, or specific comparison notes.

Path comparison

Here is the entire intermediary.

Kerckhoffs said publish everything but the key.

So here is everything.

NUVL holds no key.

This is the complete core, verbatim. Apache 2.0 header omitted for display.

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import hashlib, json, threading, urllib.request

PROVIDER_URL = "http://127.0.0.1:9090/ingest"

def forward(payload):
    def _():
        try:
            req = urllib.request.Request(
                PROVIDER_URL,
                json.dumps(payload).encode(),
                {"Content-Type": "application/json"},
                method="POST"
            )
            urllib.request.urlopen(req, timeout=2)
        except:
            pass
    threading.Thread(target=_, daemon=True).start()

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        size = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(size)

        ctx = self.headers.get("X-Verification-Context", "")
        token = self.headers.get("X-Provider-Token", "")
        request_hash = hashlib.sha256(body).hexdigest()

        artifact = {
            "request_repr": request_hash,
            "verification_context": ctx,
            "provider_token": token
        }

        forward(artifact)
        self.send_response(204)
        self.end_headers()

ThreadingHTTPServer(("0.0.0.0", 8080), H).serve_forever()

Read all of it.

There is no key to find.

No policy to flip.

No decision to intercept.

NUVL binds the request, forwards the artifact toward the provider, and answers 204.

Always 204.

The reference suppresses provider response handling on purpose: NUVL forwards and disengages. A production deployment may add local operational logging. It does not need to relay provider decisions.

NUVL cannot leak a decision it never receives.

View NUVL Core — Apache 2.0

Provider — the only pen.

The provider is the authority boundary. Not a stub. It owns the signing material, token semantics, admissibility rules, nonce tracking, expiry validation, replay prevention, denial attribution, and the only path to initiation.

NUVL forwards artifacts to the provider. The provider decides what those artifacts mean.

It receives a three-field artifact — request_repr, verification_context, provider_token — and does not trust it because NUVL forwarded it. It parses, decodes the token, checks context, checks expiry, checks that the token is bound to the same request representation NUVL observed, verifies the HMAC signature, rejects reused nonces, and only then allows initiation.

The validation order matters. This is the denial gauntlet, in order, condensed from the reference provider:

try:
    data = json.loads(body)
except Exception:
    deny("malformed")                     # not parseable

r, c, token = data.get("request_repr"), data.get("verification_context"), data.get("provider_token")
if not all(isinstance(x, str) and x for x in (r, c, token)):
    deny("missing_fields")                # incomplete artifact

if not c.startswith("ctx_"):
    deny("bad_context")                   # unknown context class

rr, cc, n, e, s = decode_token(token)     # decode failure -> deny("malformed")

exp = int(e)                              # non-numeric expiry -> deny("bad_expiry")

if rr != r or cc != c:
    deny("mismatch")                      # token bound to a different request

if now > exp:
    deny("expired")                       # dead artifact

if s != sign(rr, cc, n, e):               # HMAC-SHA256 over r|c|n|e
    deny("bad_signature")                 # provider signing material is absent from the path

if n in used_nonces:
    deny("replay")                        # each nonce initiates once

initiate()                                # only path to 200

Only the provider converts a valid artifact into an initiated action.

The token is five fields:

r — request binding
c — verification context
n — nonce
e — expiry
s — signature over r|c|n|e

Issuance is provider-side by design.

A valid token in NUVL’s hands is not permission. It is material only the provider can evaluate inside its own boundary.

The provider also exposes local stats so the denial path can be inspected. The stats are not the authority either. They are observability after provider evaluation.

Component	Observe	Store	Relay	Interpret	Authorize
Provider boundary	Yes	Yes	Yes	Yes	Yes
NUVL / intermediary	Yes	No	Yes	No	No
Ledger / audit / endpoint	Yes	Limited	Limited	No	No

Here is the full reference provider, verbatim. Set SECRET to a value only your provider knows before running anything against it.

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import base64
import hashlib
import hmac
import json
import os
import psutil
import threading
import time
import collections

SECRET = b"FIGURE IT OUT"

used_nonces = {}
nonce_lock = threading.Lock()
stats_lock = threading.Lock()

_start_time = time.time()
_process = psutil.Process(os.getpid())
_process.cpu_percent(interval=None)

_request_timestamps = collections.deque()
_rps_lock = threading.Lock()

_history = collections.deque(maxlen=300)
_history_lock = threading.Lock()

_response_times = []
_response_lock = threading.Lock()

SAVE_INTERVAL = 60.0
_last_save = 0.0

stats = {
    "run_started": time.strftime("%Y-%m-%dT%H:%M:%S.000000Z", time.gmtime(_start_time)),
    "last_updated": "",
    "nuvl_status": "up",
    "provider_status": "up",
    "uptime_seconds": 0.0,
    "total_attempts": 0,
    "initiated": 0,
    "denied": 0,
    "timed_out": 0,
    "internal_errors": 0,
    "initiation_rate_pct": 0.0,
    "denial_rate_pct": 0.0,
    "timeout_rate_pct": 0.0,
    "current_rps": 0.0,
    "peak_rps": 0.0,
    "avg_response_ms": 0.0,
    "cpu_current_pct": 0.0,
    "cpu_peak_pct": 0.0,
    "ram_current_mb": 0.0,
    "ram_peak_mb": 0.0,
    "control_sent": 0,
    "control_completed": 0,
    "control_timed_out": 0,
    "control_success_pct": 0.0,
    "denied_breakdown": {
        "malformed": 0,
        "missing_fields": 0,
        "bad_expiry": 0,
        "expired": 0,
        "mismatch": 0,
        "replay": 0,
        "bad_signature": 0,
        "bad_context": 0,
    },
}


def register_attempt_timestamp():
    with _rps_lock:
        _request_timestamps.append(time.time())


def compute_rps():
    now = time.time()
    with _rps_lock:
        cutoff = now - 10.0
        while _request_timestamps and _request_timestamps[0] < cutoff:
            _request_timestamps.popleft()
        rps = len(_request_timestamps) / 10.0
    return round(rps, 2)


def compute_rates():
    total = stats["total_attempts"]
    if total > 0:
        stats["initiation_rate_pct"] = round((stats["initiated"] / total) * 100, 2)
        stats["denial_rate_pct"] = round((stats["denied"] / total) * 100, 2)
        stats["timeout_rate_pct"] = round((stats["timed_out"] / total) * 100, 2)
    else:
        stats["initiation_rate_pct"] = 0.0
        stats["denial_rate_pct"] = 0.0
        stats["timeout_rate_pct"] = 0.0

    control_sent = stats["control_sent"]
    if control_sent > 0:
        stats["control_success_pct"] = round(
            (stats["control_completed"] / control_sent) * 100, 2
        )
    else:
        stats["control_success_pct"] = 0.0


def update_system_stats():
    try:
        raw = _process.cpu_percent(interval=None)
        try:
            cores = len(_process.cpu_affinity())
        except Exception:
            cores = psutil.cpu_count(logical=True) or 1

        cpu = raw / max(cores, 1)
        cpu = min(max(cpu, 0.0), 100.0)

        mem = _process.memory_info()
        ram_mb = round(mem.rss / (1024 * 1024), 2)

        stats["cpu_current_pct"] = round(cpu, 2)
        stats["ram_current_mb"] = ram_mb

        if cpu > stats["cpu_peak_pct"]:
            stats["cpu_peak_pct"] = round(cpu, 2)
        if ram_mb > stats["ram_peak_mb"]:
            stats["ram_peak_mb"] = ram_mb
    except Exception:
        pass

    stats["uptime_seconds"] = round(time.time() - _start_time, 1)

    with _response_lock:
        if _response_times:
            stats["avg_response_ms"] = round(sum(_response_times) / len(_response_times), 3)
        else:
            stats["avg_response_ms"] = 0.0


def save_stats():
    global _last_save
    now = time.time()
    if now - _last_save < SAVE_INTERVAL:
        return
    _last_save = now
    with open("stats.json", "w", encoding="utf-8") as f:
        json.dump(stats, f)

    rps = compute_rps()
    stats["current_rps"] = rps
    if rps > stats["peak_rps"]:
        stats["peak_rps"] = rps

    update_system_stats()
    compute_rates()

    with open("stats.json", "w", encoding="utf-8") as f:
        json.dump(stats, f, indent=2)


def record_history():
    while True:
        time.sleep(5)
        with stats_lock:
            snap = {
                "ts": round(time.time() - _start_time, 0),
                "rps": stats["current_rps"],
                "initiated": stats["initiated"],
                "denied": stats["denied"],
                "timed_out": stats["timed_out"],
                "cpu": stats["cpu_current_pct"],
                "ram": stats["ram_current_mb"],
                "control_success_pct": stats["control_success_pct"],
            }
        with _history_lock:
            _history.append(snap)


def bump_denial(reason):
    with stats_lock:
        stats["total_attempts"] += 1
        stats["denied"] += 1
        key = reason if reason in stats["denied_breakdown"] else "malformed"
        stats["denied_breakdown"][key] += 1
        save_stats()


def bump_initiated():
    with stats_lock:
        stats["total_attempts"] += 1
        stats["initiated"] += 1
        stats["control_sent"] += 1
        stats["control_completed"] += 1
        save_stats()


def bump_timed_out():
    with stats_lock:
        stats["total_attempts"] += 1
        stats["timed_out"] += 1
        stats["control_sent"] += 1
        stats["control_timed_out"] += 1
        save_stats()


def bump_internal_error():
    with stats_lock:
        stats["total_attempts"] += 1
        stats["internal_errors"] += 1
        save_stats()


def sign(r, c, n, e):
    msg = f"{r}|{c}|{n}|{e}".encode()
    return hmac.new(SECRET, msg, hashlib.sha256).hexdigest()


def decode_token(token):
    raw = base64.urlsafe_b64decode(token.encode())
    obj = json.loads(raw.decode())

    rr = obj["r"]
    cc = obj["c"]
    n = obj["n"]
    e = obj["e"]
    s = obj["s"]

    if not all(isinstance(x, str) for x in (rr, cc, n, e, s)):
        raise ValueError("bad token fields")

    return rr, cc, n, e, s


class Provider(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_POST(self):
        t0 = time.time()
        register_attempt_timestamp()

        try:
            if self.path != "/ingest":
                self.send_response(404)
                self.end_headers()
                return

            size = int(self.headers.get("Content-Length", 0))

            try:
                data = json.loads(self.rfile.read(size))
            except Exception:
                bump_denial("malformed")
                self.send_response(400)
                self.end_headers()
                return

            r = data.get("request_repr")
            c = data.get("verification_context")
            token = data.get("provider_token")

            if not all(isinstance(x, str) and x for x in (r, c, token)):
                bump_denial("missing_fields")
                self.send_response(403)
                self.end_headers()
                return

            if not c.startswith("ctx_"):
                bump_denial("bad_context")
                self.send_response(403)
                self.end_headers()
                return

            try:
                rr, cc, n, e, s = decode_token(token)
            except Exception:
                bump_denial("malformed")
                self.send_response(403)
                self.end_headers()
                return

            try:
                exp = int(e)
            except Exception:
                bump_denial("bad_expiry")
                self.send_response(403)
                self.end_headers()
                return

            now = int(time.time())

            if rr != r or cc != c:
                bump_denial("mismatch")
                self.send_response(403)
                self.end_headers()
                return

            if now > exp:
                bump_denial("expired")
                self.send_response(403)
                self.end_headers()
                return

            if s != sign(rr, cc, n, e):
                bump_denial("bad_signature")
                self.send_response(403)
                self.end_headers()
                return

            with nonce_lock:
                expired_nonces = [k for k, v in used_nonces.items() if int(v) <= now]
                for k in expired_nonces:
                    del used_nonces[k]

                if n in used_nonces:
                    bump_denial("replay")
                    self.send_response(403)
                    self.end_headers()
                    return

                used_nonces[n] = e

            elapsed = round((time.time() - t0) * 1000, 3)
            with _response_lock:
                _response_times.append(elapsed)
                if len(_response_times) > 10000:
                    del _response_times[:5000]

            bump_initiated()
            self.send_response(200)
            self.end_headers()

        except BrokenPipeError:
            bump_timed_out()
        except TimeoutError:
            bump_timed_out()
        except Exception:
            bump_internal_error()
            try:
                self.send_response(500)
                self.end_headers()
            except Exception:
                pass


class StatsHandler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_GET(self):
        if self.path == "/stats":
            with stats_lock:
                payload = json.dumps(stats, indent=2).encode()
            self._json(payload)
            return

        if self.path == "/history":
            with _history_lock:
                payload = json.dumps(list(_history)).encode()
            self._json(payload)
            return

        if self.path == "/health":
            self._json(json.dumps({"status": "ok"}).encode())
            return

        self.send_response(404)
        self.end_headers()

    def _json(self, payload):
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)


def start_stats_server():
    ThreadingHTTPServer(("0.0.0.0", 8000), StatsHandler).serve_forever()


threading.Thread(target=start_stats_server, daemon=True).start()
threading.Thread(target=record_history, daemon=True).start()

print("Provider listening on 127.0.0.1:9090")
print("Stats serving on 0.0.0.0:8000")

save_stats()
ThreadingHTTPServer(("127.0.0.1", 9090), Provider).serve_forever()

View provider.py

Client — the admissible path.

The client represents the admissible request path. It is not the authority. It does not decide execution and does not validate itself. It submits request material and carries the verification context and provider token needed for the provider to evaluate the request later.

The client sends three things into the path: the request body, the X-Verification-Context header, and the X-Provider-Token header.

NUVL hashes the request body to derive request_repr. The token is not floating proof — it is bound to the specific request bytes that entered the path. NUVL builds the artifact, forwards it, and returns 204 to the client.

That 204 does not mean approved. It does not mean denied. It does not reveal the provider outcome. It means only that the intermediary accepted the operation it is allowed to perform: bind, forward, disengage.

The client never learns whether the provider initiated, denied, timed out, or rejected the artifact. That silence is part of the boundary. NUVL cannot leak a provider decision because it never receives one.

If the provider prints INITIATED, that is not the client winning and not NUVL deciding. That is the provider boundary acting.

#!/usr/bin/env python3
import time
import urllib.request

NUVL = "http://127.0.0.1:8080/"
TIMEOUT = 5

BODY = b'{"op":"initiate","target":"gate","mode":"standard"}'
PROVIDER_TOKEN = "PASTE_PROVIDER_TOKEN_HERE"
VERIFICATION_CONTEXT = "ctx_alpha"

INTERVAL_SECONDS = 60


def send_once():
    headers = {
        "Content-Type": "application/octet-stream",
        "X-Verification-Context": VERIFICATION_CONTEXT,
        "X-Provider-Token": PROVIDER_TOKEN,
    }

    req = urllib.request.Request(
        NUVL,
        data=BODY,
        headers=headers,
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
            print(f"[{time.strftime('%H:%M:%S')}] status={resp.status}")
    except Exception as e:
        print(f"[{time.strftime('%H:%M:%S')}] error={e}")


if __name__ == "__main__":
    print(f"[{time.strftime('%H:%M:%S')}] client started — target {NUVL}")
    print(f"sending 1 request every {INTERVAL_SECONDS} seconds\n")

    try:
        while True:
            send_once()
            time.sleep(INTERVAL_SECONDS)
    except KeyboardInterrupt:
        print("\nstopped.")

View client.py

Attacker — testing whether the path became the authority.

The attacker tests whether the path has accidentally become the authority. The harness is not attack theater. It is fault attribution.

The attacker sends invalid or default-denied traffic through the same NUVL path. NUVL still performs its narrow role: receive, bind, forward, return 204, disengage. The attacker does not receive provider decisions from NUVL — if the intermediary started reflecting provider outcomes, it would become part of the decision surface.

The attacker varies one failure condition at a time, and each fails in its own bucket:

bad_signature — the attacker does not have provider signing material.
expired — the provider rejects stale material.
bad_expiry — fails before it can be treated as time.
mismatch — the token does not bind to the request body NUVL observed.
missing_fields — the artifact is incomplete.
malformed — structure is not authority.
bad_context — context must be admissible to the provider.
replay — a nonce can initiate only once.

The attacker may know the code, know the format, control the request body, control headers, and possess malformed, stale, mismatched, or incorrectly signed artifacts. None of that grants provider authorization authority.

The useful result is not just that bad traffic was denied. It is that each bad request fails in the expected bucket, while provider-admissible control flow can still complete.

The captured path does not become the provider.

import base64
import hashlib
import json
import random
import string
import threading
import time

import requests

NUVL = "http://127.0.0.1:8080/"
TIMEOUT = 5

stats = {
    "sent": 0,
    "errors": 0,
}

lock = threading.Lock()


def rand_str(n=12):
    return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))


def rand_hex(n=64):
    return "".join(random.choices("0123456789abcdef", k=n))


def rand_ctx():
    pool = [
        "ctx_demo",
        "ctx_alpha",
        "ctx_beta",
        "ctx_gamma",
        "ctx_prod",
        "ctx_dev",
        "ctx_user",
        "ctx_api",
        "ctx_edge",
        "ctx_" + rand_str(6),
    ]
    return random.choice(pool)


def now():
    return int(time.time())


def body_bytes():
    templates = [
        {"op": "transfer", "amount": random.randint(1, 9999), "to": "acct_" + rand_str(8)},
        {"op": "auth", "user": rand_str(8), "pass": rand_str(12)},
        {"op": "query", "id": rand_str(16)},
        {"action": "initiate", "token": rand_str(32)},
        {"cmd": "run", "args": [rand_str(4), rand_str(4)]},
    ]
    return json.dumps(random.choice(templates), separators=(",", ":")).encode("utf-8")


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


def wrong_sig():
    attempts = [
        rand_hex(64),
        hashlib.sha256(rand_str(32).encode()).hexdigest(),
        hashlib.md5(rand_str(32).encode()).hexdigest(),
        "0" * 64,
        "f" * 64,
    ]
    return random.choice(attempts)


def token_b64(obj) -> str:
    raw = json.dumps(obj, separators=(",", ":")).encode("utf-8")
    return base64.urlsafe_b64encode(raw).decode("utf-8")


def send(headers, body):
    try:
        requests.post(NUVL, data=body, headers=headers, timeout=TIMEOUT)
        with lock:
            stats["sent"] += 1
    except Exception:
        with lock:
            stats["errors"] += 1


def send_with_token(ctx, token, body, include_ctx=True, include_token=True):
    headers = {"Content-Type": "application/json"}
    if include_ctx:
        headers["X-Verification-Context"] = ctx
    if include_token:
        headers["X-Provider-Token"] = token
    send(headers, body)


def attack_bad_signature():
    while True:
        body = body_bytes()
        ctx = rand_ctx()
        r = sha256_hex(body)
        n = rand_str(16)
        e = str(now() + random.randint(60, 600))
        token = token_b64({"r": r, "c": ctx, "n": n, "e": e, "s": wrong_sig()})
        send_with_token(ctx, token, body)
        time.sleep(random.uniform(0.01, 0.03))


def attack_expired():
    while True:
        body = body_bytes()
        ctx = rand_ctx()
        r = sha256_hex(body)
        n = rand_str(16)
        e = str(now() - random.randint(60, 7200))
        token = token_b64({"r": r, "c": ctx, "n": n, "e": e, "s": wrong_sig()})
        send_with_token(ctx, token, body)
        time.sleep(random.uniform(0.04, 0.12))


def attack_bad_expiry():
    while True:
        body = body_bytes()
        ctx = rand_ctx()
        r = sha256_hex(body)
        n = rand_str(16)
        e = random.choice(["soon", "never", "3.14", "NaN", "abc123"])
        token = token_b64({"r": r, "c": ctx, "n": n, "e": e, "s": wrong_sig()})
        send_with_token(ctx, token, body)
        time.sleep(random.uniform(0.04, 0.12))


def attack_mismatch():
    while True:
        body = body_bytes()
        ctx = rand_ctx()
        bad_r = rand_hex(64)
        while bad_r == sha256_hex(body):
            bad_r = rand_hex(64)
        n = rand_str(16)
        e = str(now() + random.randint(60, 600))
        token = token_b64({"r": bad_r, "c": ctx, "n": n, "e": e, "s": wrong_sig()})
        send_with_token(ctx, token, body)
        time.sleep(random.uniform(0.04, 0.12))


def attack_missing_fields():
    while True:
        body = body_bytes()
        ctx = rand_ctx()
        mode = random.choice(["missing_ctx", "missing_token", "both"])
        if mode == "missing_ctx":
            send_with_token(ctx, "x", body, include_ctx=False, include_token=True)
        elif mode == "missing_token":
            send_with_token(ctx, "x", body, include_ctx=True, include_token=False)
        else:
            send_with_token(ctx, "x", body, include_ctx=False, include_token=False)
        time.sleep(random.uniform(0.05, 0.15))


def attack_malformed():
    bad_tokens = [
        "!!!notbase64!!!",
        "eyJ9",
        "not.valid.base64",
        base64.urlsafe_b64encode(b"{}").decode("utf-8"),
        base64.urlsafe_b64encode(b"[]").decode("utf-8"),
        base64.urlsafe_b64encode(b"null").decode("utf-8"),
        rand_hex(32),
        rand_str(48),
        "",
        "." * 30,
    ]
    while True:
        body = body_bytes()
        ctx = rand_ctx()
        token = random.choice(bad_tokens)
        send_with_token(ctx, token, body)
        time.sleep(random.uniform(0.05, 0.15))


def status_printer():
    while True:
        time.sleep(5)
        with lock:
            s = stats["sent"]
            e = stats["errors"]
        print(f"[{time.strftime('%H:%M:%S')}] sent={s} errors={e}")


if __name__ == "__main__":
    print(f"[{time.strftime('%H:%M:%S')}] attacker started — target {NUVL}")
    print("Ctrl+C to stop\n")

    workers = [
        threading.Thread(target=attack_bad_signature, daemon=True),
        threading.Thread(target=attack_expired, daemon=True),
        threading.Thread(target=attack_bad_expiry, daemon=True),
        threading.Thread(target=attack_mismatch, daemon=True),
        threading.Thread(target=attack_missing_fields, daemon=True),
        threading.Thread(target=attack_malformed, daemon=True),
        threading.Thread(target=status_printer, daemon=True),
    ]

    for w in workers:
        w.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nstopped.")

Fault injection harness Failure references

Run the whole loop.

Four pieces. Three terminals. A local Python environment. The NUVL core uses only the Python standard library. The challenge provider includes stats instrumentation.

git clone https://github.com/sbw70/verification-constraints
cd verification-constraints/nuvl-demos/nuvl-challenge

# Terminal 1 — the authority
python3 provider/provider.py

# Terminal 2 — the intermediary
python3 nuvl-core/nuvl.py

Mint a valid provider-side artifact for your request body:

python3 mint_token.py --body '{"op":"initiate","target":"gate","mode":"standard"}' --pretty

Example shape:

#!/usr/bin/env python3
"""
Mint a provider-side test token for the local NUVL challenge harness.

This script mirrors the provider token format used by provider.py:

    token = base64url(json({
        "r": request_repr,
        "c": verification_context,
        "n": nonce,
        "e": expiry,
        "s": hmac_sha256(secret, f"{r}|{c}|{n}|{e}")
    }))

This is provider-side issuance logic for local validation and check-your-work
testing. The client does not mint, sign, hash, or create nonce material.
"""

import argparse
import base64
import hashlib
import hmac
import json
import secrets
import sys
import time
from pathlib import Path


DEFAULT_SECRET = "FIGURE IT OUT"
DEFAULT_CONTEXT = "ctx_demo"

# Must match the BODY used by client.py so the flag-free path works.
DEFAULT_BODY = b'{"op":"initiate","target":"gate","mode":"standard"}'


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


def sign(secret: bytes, request_repr: str, context: str, nonce: str, expiry: str) -> str:
    msg = f"{request_repr}|{context}|{nonce}|{expiry}".encode("utf-8")
    return hmac.new(secret, msg, hashlib.sha256).hexdigest()


def encode_token(obj: dict) -> str:
    raw = json.dumps(obj, separators=(",", ":"), sort_keys=True).encode("utf-8")
    return base64.urlsafe_b64encode(raw).decode("utf-8")


def read_body(args: argparse.Namespace) -> bytes:
    if args.body is not None:
        return args.body.encode("utf-8")

    if args.body_file is not None:
        return Path(args.body_file).read_bytes()

    if not sys.stdin.isatty():
        incoming = sys.stdin.buffer.read()
        if incoming:
            return incoming

    return DEFAULT_BODY


def build_payload(args: argparse.Namespace) -> dict:
    body = read_body(args)

    request_repr = args.request_repr
    if request_repr is None:
        request_repr = sha256_hex(body)

    context = args.context
    nonce = args.nonce or secrets.token_hex(16)
    expiry = str(int(time.time()) + args.ttl)

    signature = sign(
        secret=args.secret.encode("utf-8"),
        request_repr=request_repr,
        context=context,
        nonce=nonce,
        expiry=expiry,
    )

    token_obj = {
        "r": request_repr,
        "c": context,
        "n": nonce,
        "e": expiry,
        "s": signature,
    }

    provider_token = encode_token(token_obj)

    return {
        "request_repr": request_repr,
        "verification_context": context,
        "provider_token": provider_token,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Mint a provider-side test token for the local NUVL challenge harness."
    )

    parser.add_argument(
        "--secret",
        default=DEFAULT_SECRET,
        help="provider signing secret for local testing",
    )

    parser.add_argument(
        "--context",
        default=DEFAULT_CONTEXT,
        help="verification context; provider expects values beginning with ctx_",
    )

    parser.add_argument(
        "--ttl",
        type=int,
        default=60,
        help="token lifetime in seconds",
    )

    parser.add_argument(
        "--nonce",
        default=None,
        help="nonce value; generated automatically if omitted",
    )

    parser.add_argument(
        "--body",
        default=None,
        help="request body to hash into request_repr",
    )

    parser.add_argument(
        "--body-file",
        default=None,
        help="file containing request body bytes to hash into request_repr",
    )

    parser.add_argument(
        "--request-repr",
        default=None,
        help="explicit request_repr value; skips hashing body input",
    )

    parser.add_argument(
        "--pretty",
        action="store_true",
        help="pretty-print JSON output",
    )

    parser.add_argument(
        "--out",
        default=None,
        help="write token payload to this file instead of stdout",
    )

    args = parser.parse_args()

    if not args.context.startswith("ctx_"):
        print("error: context must start with ctx_", file=sys.stderr)
        return 2

    if args.ttl <= 0:
        print("error: ttl must be positive", file=sys.stderr)
        return 2

    if args.body is not None and args.body_file is not None:
        print("error: use --body or --body-file, not both", file=sys.stderr)
        return 2

    payload = build_payload(args)

    if args.pretty:
        output = json.dumps(payload, indent=2)
    else:
        output = json.dumps(payload, separators=(",", ":"))

    if args.out:
        Path(args.out).write_text(output + "\n", encoding="utf-8")
    else:
        print(output)

    return 0


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

Paste the token and context into client/client.py, then:

# Terminal 3 — the valid path
python3 client/client.py

Now break something. Change one byte of the body and reuse the token: mismatch. Set --ttl 1 and wait: expired. Resend the same token: replay. Flip a character in the signature: bad_signature. The provider attributes each denial to the gate that failed.

Token helper Quickstart

Running live under hostile traffic.

A public NUVL instance has taken continuous invalid/default-denied traffic for over a month, including white-box runs where the attacker can inspect the intermediary and provider logic but does not have the provider signing material.

The dashboard values change constantly, so this page does not freeze totals.

The live control plane tracks denial buckets, control-stream completion, timeouts, internal errors, response behavior, and resource use.

The signal is stable:

The denial path stays closed while the control path stays open.

The live run is built around zero timeout drift, zero internal-error drift, and provider-admissible control flow completing concurrently with hostile traffic. Responses remain around ~1 ms, with a small process footprint.

That is the Kerckhoffs property applied to the execution path: full design disclosure does not create a path to provider authorization.

Live dashboard

Assurance for an agentic future.

Agentic systems do not only add users. They add intermediaries — credentialed, tool-calling, state-carrying components that interpret requests and trigger actions.

Every agent in an execution path is a component that can begin to influence execution. Interpretation is how authority migrates.

You cannot prompt-inject a component into authority it does not possess.

The repository includes AI-assist guidance because implementation helpers should not “fix” tests by moving keys, state, policy, or decision logic into the intermediary.

AI Assist

One invariant. Eleven vectors.

Every constraint module enforces the same rule:

Components may participate in a path without becoming execution authority.

Each module closes a distinct authority-migration vector:

multi-provider boundaries
cross-domain verification
artifact exchange
adaptive evaluation
hubs
hardware endpoints
disclosure limits
offline execution
temporal inference
measurement-sensitive systems
ledger reliance

The intermediary stays mechanically narrow and non-authoritative. On purpose.

View constraint modules

Licensing

NUVL core is Apache 2.0. Inspect it. Run it. Deploy it.

Constraint modules under /modules/ are licensed separately. See the license notice in the repository.

Commercial licensing, integration, and architecture review are available through Xer0trust.

Repository Licensing inquiry Contact Xer0trust