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 to the intermediary environment. Full memory. Complete source.

This assumes the provider boundary and signing material remain separately protected. The local walkthrough runs both processes on one machine to demonstrate the request flow; it does not isolate the provider from someone with root access to that machine.

What does the attacker now hold?

The provider signing key — no. The neutral relay does not hold it.
The power to change provider authorization policy — no. That remains at the provider boundary.
Retained session or authorization state — none in this reference. In-flight request material is still visible.
Provider HTTP responses — observable at the captured relay. The unmodified reference does not relay them to the caller.
Valid artifacts passing through the relay — yes. An attacker can submit an unused artifact while it remains valid, within its provider-established limits.

The attacker can observe, alter, drop, delay, or resubmit traffic. Those capabilities matter. They do not supply the provider's signing key or permit the attacker to mint or enlarge provider-issued authority.

Relay control does not become provider control. The provider must still validate every artifact, including artifacts arriving through a compromised relay.

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.

The Path Comparison notes document the two instrumented paths, measurement definitions, run scope, deployment context, and limits of the comparison.

Path comparison

Where enforcement runs.

The HTTP walkthrough below uses NUVL as a neutral relay. Other implementations place enforcement at a separate boundary or on the endpoint. In each case, the provider establishes the authority and its limits.

Neutral relay. The reference hashes the request body and forwards an artifact. The provider checks admissibility and keeps used nonces in RAM.

Separate enforcement boundary. The provider issues bounded authority. A separate component verifies and enforces it using the configured trust anchor, retaining consumption state when required.

Endpoint-local enforcement. The device performs those enforcement duties before issuing its command. An implementation requiring replay protection across restarts must retain the consumed state durably.

Stateless applies to the neutral relay. Enforcement may require state. Keeping that state does not give an enforcement component the right to originate or enlarge provider-issued authority.

Here is the entire intermediary.

Kerckhoffs said publish everything but the key.

So here is everything.

NUVL holds no key.

This is the complete neutral core, including its Apache 2.0 header.

#!/usr/bin/env python3
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# SPDX-License-Identifier: Apache-2.0

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 provider authorization decision is made here.

For a POST that reaches the response step, NUVL hashes the body, starts an asynchronous forwarding attempt, and answers 204.

The 204 is independent of the provider's decision. It does not confirm delivery, admission, or execution. Other HTTP methods and requests that fail before the response step do not share that guarantee.

The HTTP forwarding call receives a provider response or an error. The reference discards the returned response and suppresses exceptions; it does not relay the provider's status or response body to the original caller.

An application that needs the operation's outcome must obtain it through a separately defined provider or application result path.

View NUVL Core — Apache 2.0

Provider — the only pen.

The provider is the authority boundary in this HTTP demonstration. It owns the signing material, token semantics, admissibility rules, nonce tracking, expiry validation, replay prevention, and denial attribution. Successful validation records an admission in the initiated counter.

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 the artifact, checks required fields and context, decodes the token and expiry, checks the request binding and deadline, verifies the HMAC, then checks and consumes the nonce under a lock before recording admission.

The validation order matters. This outline shows the checks in order; it is not a standalone program. The complete runnable provider follows:

# Validation outline: deny(reason) records a denial and ends the request.
try:
    data = json.loads(body)
except Exception:
    deny("malformed")

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")

if not c.startswith("ctx_"):
    deny("bad_context")

rr, cc, n, e, s = decode_token(token)  # decode failure -> deny("malformed")
exp = int(e)                           # conversion failure -> deny("bad_expiry")
now = int(time.time())

if rr != r or cc != c:
    deny("mismatch")

if now >= exp:
    deny("expired")

if s != sign(rr, cc, n, e):            # HMAC-SHA256 over r|c|n|e
    deny("bad_signature")

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:
        deny("replay")
    used_nonces[n] = e                 # consume in this process's RAM

bump_initiated()                       # records admission; HTTP 200 follows

Only the provider records admission. This demonstration does not actuate the gate named in the request or confirm completion of a physical operation. Used nonces are retained in RAM and are lost when this provider process restarts.

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 carries provider-established authority. Possessing it can permit submission of the bound request while its conditions remain satisfied. It does not permit the relay to change the authorized request, issue new tokens, or enlarge the grant.

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. For local evaluation, it and mint_token.py share the published fixture secret FIGURE IT OUT. If you change SECRET before starting your local provider, pass that same value to the token helper with --secret. The fixture value is public test material, not a deployment secret.

#!/usr/bin/env python3
# NUVL Challenge Demonstration - Limited Evaluation Permission
#
# Permission to run and modify this file for local, noncommercial evaluation
# is governed by "Challenge Demonstration Evaluation" in
# module-license-notice/README.md at the repository root.
# Uses outside that permission require a separate written agreement.
# This file is not licensed under Apache 2.0.
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 provider checks that this representation and the context match the signed token. The neutral reference builds the artifact, attempts forwarding asynchronously, and returns 204 for a handled POST.

That 204 reports the intermediary response, not provider approval, denial, or successful delivery. Read the local provider statistics to observe the provider result in this demonstration.

The client does not receive the provider outcome through this relay response. The HTTP forwarding call does receive a provider response or error, but the reference does not pass it back to the client.

The provider increments initiated when it admits a request; it does not print an INITIATED message. That counter records admission, not completed physical execution.

#!/usr/bin/env python3
# NUVL Challenge Demonstration - Limited Evaluation Permission
#
# Permission to run and modify this file for local, noncommercial evaluation
# is governed by "Challenge Demonstration Evaluation" in
# module-license-notice/README.md at the repository root.
# Uses outside that permission require a separate written agreement.
# This file is not licensed under Apache 2.0.
import argparse
import json
import os
import time
from pathlib import Path

import requests

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

BODY = b'{"op":"initiate","target":"gate","mode":"standard"}'

# Must match the context inside the provider-issued token.
# mint_token.py defaults to ctx_demo, so the naive path works flag-free.
VERIFICATION_CONTEXT = "ctx_demo"

INTERVAL_SECONDS = 60


def load_provider_token(token_file: str | None, token_value: str | None) -> str:
    """
    Load a provider-issued token.

    Accepted formats:
      1. raw token string
      2. JSON object from mint_token.py containing "provider_token"

    This client does not mint, sign, hash, or create nonce material.
    """

    if token_value:
        return token_value.strip()

    if not token_file:
        raise RuntimeError("missing provider token: use --token or --token-file")

    raw = Path(token_file).read_text(encoding="utf-8").strip()

    try:
        obj = json.loads(raw)
        token = obj.get("provider_token")
        if isinstance(token, str) and token:
            return token
    except Exception:
        pass

    if not raw:
        raise RuntimeError(f"empty provider token file: {token_file}")

    return raw


def send_once(token_file: str | None, token_value: str | None) -> bool:
    try:
        provider_token = load_provider_token(token_file, token_value)
    except Exception as e:
        print(f"[{time.strftime('%H:%M:%S')}] error={e}")
        return False

    headers = {
        "Content-Type": "application/octet-stream",
        "X-Verification-Context": VERIFICATION_CONTEXT,
        "X-Provider-Token": provider_token,
    }

    try:
        r = requests.post(NUVL, data=BODY, headers=headers, timeout=TIMEOUT)
        print(f"[{time.strftime('%H:%M:%S')}] status={r.status_code}")
        return True
    except Exception as e:
        print(f"[{time.strftime('%H:%M:%S')}] error={e}")
        return False


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Send a request to local NUVL using a provider-issued token."
    )

    parser.add_argument(
        "--token",
        default=os.environ.get("PROVIDER_TOKEN"),
        help="provider-issued token; useful for one request",
    )

    parser.add_argument(
        "--token-file",
        default=os.environ.get("PROVIDER_TOKEN_FILE"),
        help="file containing either a raw token or mint_token.py JSON output",
    )

    parser.add_argument(
        "--once",
        action="store_true",
        help="send one request and exit",
    )

    args = parser.parse_args()

    print(f"[{time.strftime('%H:%M:%S')}] client started — target {NUVL}")
    print(f"context={VERIFICATION_CONTEXT}")

    if args.once:
        return 0 if send_once(args.token_file, args.token) else 2

    if args.token and not args.token_file:
        print(
            "warning: static --token/PROVIDER_TOKEN will be replayed if looped; "
            "provider should issue a fresh token per request"
        )

    print(f"sending 1 request every {INTERVAL_SECONDS} seconds")
    print("token file is re-read before every request\n")

    try:
        while True:
            send_once(args.token_file, args.token)
            time.sleep(INTERVAL_SECONDS)
    except KeyboardInterrupt:
        print("\nstopped.")
        return 0


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

View client.py

Attacker — testing whether the path became the authority.

This local harness sends invalid traffic through the same NUVL path and lets you inspect which provider checks reject it. It targets 127.0.0.1:8080. It is a local evaluation variant, distinct from the live challenge harness in the repository.

NUVL handles these POST requests in its usual way: receive, hash, attempt forwarding, and return 204. The harness does not receive provider decisions through that response. Inspect denied_breakdown in the local provider statistics to see the results.

Its six workers exercise these first-failure categories. Several requests contain more than one defect; the provider records the first failed check in its validation order:

bad_signature — the token carries an invalid HMAC.
expired — the deadline has passed; this check occurs before signature verification.
bad_expiry — the expiry cannot be parsed as an integer.
mismatch — the token's request representation differs from the body hash.
missing_fields — the context or token is empty or absent.
malformed — the token cannot be decoded into the required fields.

This harness does not exercise bad_context or valid-token replay. Its malformed-token worker can also reach missing_fields when it sends an empty token. The walkthrough below tests replay by submitting a genuinely valid token twice before expiry.

The harness reports requests sent and client-side errors. Those are not provider denial counts. Read the provider counters to establish which requests arrived and which checks rejected them.

These are local denial checks. Sending invalid traffic does not by itself demonstrate every consequence of relay compromise.

#!/usr/bin/env python3
# NUVL Challenge Demonstration - Limited Evaluation Permission
#
# Permission to run and modify this file for local, noncommercial evaluation
# is governed by "Challenge Demonstration Evaluation" in
# module-license-notice/README.md at the repository root.
# Uses outside that permission require a separate written agreement.
# This file is not licensed under Apache 2.0.
#
# Local evaluation variant: traffic is sent only to the local NUVL demo.
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.")

Live challenge harness source Failure references

Run the whole loop.

Four pieces. Three terminals. Start with Git and Python 3.10 or later, including pip and venv. The NUVL core uses only the Python standard library. This demo's provider and client also require psutil and requests.

Use a local test environment with ports 8000, 8080, and 9090 available. Every terminal must be in the same verification-constraints/nuvl-demos/nuvl-challenge directory. The provider writes stats.json there.

The marked demonstration files permit local, noncommercial evaluation under the challenge evaluation terms. The neutral core retains its Apache 2.0 license.

Linux / macOS commands

1. Download and install the demo dependencies. These commands create a virtual environment inside the challenge directory.

git clone https://github.com/sbw70/verification-constraints
cd verification-constraints/nuvl-demos/nuvl-challenge
python3 -m venv .venv
.venv/bin/python -m pip install requests psutil

2. Terminal 1: start the provider. Keep it running.

.venv/bin/python -u provider/provider.py

3. Terminal 2: start NUVL. Open another terminal, navigate to the same challenge directory, and keep this process running.

.venv/bin/python nuvl-core/nuvl.py

4. Terminal 3: mint a token and send one request. Navigate to the same challenge directory first. The helper's default body and context match the client. The token is written to a file; no source editing is needed.

If you changed SECRET in the provider, pass the same value here with --secret, or the token will be rejected as bad_signature.

.venv/bin/python mint_token.py --ttl 300 --out token.json
.venv/bin/python client/client.py --token-file token.json --once

5. Repeat the request using the same token. Run this in Terminal 3 within five minutes of minting it.

.venv/bin/python client/client.py --token-file token.json --once

Windows PowerShell commands

1. Download and install the demo dependencies. The Python launcher must select Python 3.10 or later. These commands use the virtual environment directly, without an activation script.

git clone https://github.com/sbw70/verification-constraints
cd verification-constraints/nuvl-demos/nuvl-challenge
py -3 -m venv .venv
.venv\Scripts\python.exe -m pip install requests psutil

2. Terminal 1: start the provider. Keep it running.

.venv\Scripts\python.exe -u provider/provider.py

3. Terminal 2: start NUVL. Open another terminal, navigate to the same challenge directory, and keep this process running.

.venv\Scripts\python.exe nuvl-core/nuvl.py

If you changed SECRET in the provider, pass the same value here with --secret, or the token will be rejected as bad_signature.

.venv\Scripts\python.exe mint_token.py --ttl 300 --out token.json
.venv\Scripts\python.exe client/client.py --token-file token.json --once

5. Repeat the request using the same token. Run this in Terminal 3 within five minutes of minting it.

.venv\Scripts\python.exe client/client.py --token-file token.json --once

Read the result at the provider. Open your local provider statistics on the machine running the demo and refresh after each request. With a fresh provider process, the first request raises initiated to 1. Reusing the same unexpired token leaves it at 1, raises denied to 1, and raises denied_breakdown.replay to 1.

The client can print status=204 for both requests. That response comes from NUVL and does not report the provider's decision. In this demo, initiated records provider admission.

If the token has expired, the provider records expired instead of replay. Mint a new token to begin another trial. This provider keeps used nonces in RAM; restarting it clears that replay history.

Optional: run the local denial harness.

To run the six-worker example shown above, copy it into attacker/attacker.py in your local clone, replacing that file's contents locally. Keep the provider and NUVL running. This local variant targets 127.0.0.1; the repository's original attacker file targets the live challenge and is a different harness.

In Terminal 3 on Linux or macOS:

.venv/bin/python attacker/attacker.py

Or in Terminal 3 on Windows PowerShell:

.venv\Scripts\python.exe attacker/attacker.py

Refresh the local provider statistics while the harness runs. The six listed denial counters should increase, with no additional admissions from this invalid traffic. Stop the harness with Ctrl+C, then stop the two servers when finished.

Token helper source Client source Challenge documentation

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.

Metric definitions, measurement scope, traffic provenance, run history, and known limitations are documented in the Live Dashboard notes.

The signal is stable:

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

The live run is built around zero timeout drift, stable internal-error behavior, 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