How to write runtime monitors and test oracles

This guide shows two ways to use a proved Lean specification at run time, when the code being checked is not Lean:

• A runtime monitor is fed the events of a running system, one at a time. It is proved to accept exactly the event traces a protocol allows, so every violation it reports is a real one.
• A test oracle is a proved reference implementation. Code written by hand, in any language, is tested against it rather than trusted on its own.

The worked example is lungo-rs/examples/oracle-monitor, an access-control policy with a session protocol.

Write the protocol

A protocol (Lungo.Trace.Protocol ε) is the set of event traces a component may take part in, closed under prefixes. Its executable side is a reducer (Lungo.Trace.Reducer ε σ): a start state and a step that accepts an event, giving the next state, or refuses it.

import Lungo

namespace Access

open Lungo Lungo.Trace

inductive Event where
  | signIn (user role : String)
  | access (user action : String)
  | signOut (user : String)

/-- Who is signed in, with their role. -/
structure Session where
  users : List (String × String)

/-- The session after `e`; `none` when `e` breaks the protocol. -/
def next (policy : List Rule) (s : Session) (e : Event) : Option Session := ...

/-- No one signed in: the session a host starts monitoring from. -/
def start : Session := ⟨[]⟩

/-- The session protocol under `policy`. -/
@[lungo_spec "lungo.protocol"]
noncomputable def Sessions (policy : List Rule) : Protocol Event :=
  (Reducer.mk start (next policy)).protocol

A protocol is a logical object, not code to run, so Sessions is noncomputable. lungo never exports it, and @[lungo_spec] registers it as a specification that claims can cite.

Build the monitor

A monitor (Lungo.Monitor ε σ) has a start state and a step : σ → ε → Except String σ. A step gives the next state, or the reason the event is a violation. Monitor.run feeds a whole trace, Monitor.Accepts m t says m accepts trace t from its start, and Monitor.Sound m p says m accepts exactly the traces protocol p allows.

Build monitors with these combinators. Each one comes with its soundness theorem:

The example monitors its reducer:

/-- Why `e` breaks the protocol in session `s`. -/
def reason (s : Session) (e : Event) : String := ...

private def monitor (policy : List Rule) : Monitor Event Session :=
  Monitor.ofReducer ⟨start, next policy⟩ reason

/-- The session after `e`, or the violation it is. -/
def observe (policy : List Rule) (s : Session) (e : Event) : Except String Session :=
  (monitor policy).step s e

The host needs the start state and the step as functions, so the exports are start and observe. The Monitor value itself has no use outside Lean, so it is private.

Claim it is sound

State the claim about the exported functions, under the relation lungo.monitors:

@[lungo_claim "lungo.monitors" subject start observe spec Sessions]
theorem observe_sound (policy : List Rule) :
    (Monitor.mk start (observe policy)).Sound (Sessions policy) :=
  Monitor.ofReducer_sound (Reducer.mk start (next policy)) reason

A lungo.monitors claim must state Lungo.Monitor.Sound m p, with m mentioning every subject. lungo checks this shape when the attribute is applied, and again when it reads the records. To fail the build if the claim is removed or rests on sorry, require it:

# lungo.toml
[assurance]
require-claims = ["Access.observe"]

(Builder::require_claims("Access.observe") in build.rs.)

Feed it events from the host

The host keeps the session and calls observe with each event. What to do about a violation (log it, reject the request, alert someone) is the host’s decision; the proof only guarantees that it is one.

Rust

let mut session = access::start();
for event in events {
    match access::observe(policy.clone(), session.clone(), event) {
        Ok(next) => session = next,
        Err(reason) => eprintln!("violation: {reason}"),
    }
}

Python

session = access.start()
for event in events:  # access.EventSignIn("ada", "editor"), access.EventAccess("ada", "write"), …
    verdict = access.observe(policy, session, event)
    if isinstance(verdict, lungo_py.Ok):
        session = verdict.value
    else:
        log.warning("violation: %s", verdict.error)

Go

session, err := access.Start()
if err != nil {
	return err
}
for _, e := range events { // access.EventSignIn{User: "ada", Role: "editor"}, …
	verdict, err := access.Observe(policy, session, e)
	if err != nil {
		return err
	}
	if verdict.Ok {
		session = verdict.Value
	} else {
		log.Printf("violation: %s", verdict.Error)
	}
}

TypeScript

let session = p.start();
for (const e of events) { // { kind: "signIn", user: "ada", role: "editor" }, …
  const verdict = p.observe(policy, session, e);
  if (verdict.ok) session = verdict.value;
  else console.warn(`violation: ${verdict.error}`);
}

Swift

var session = try start()
for e in events { // .signIn(user: "ada", role: "editor"), …
    switch try observe(policy, session, e) {
    case .ok(let next): session = next
    case .error(let reason): print("violation: \(reason)")
    }
}

C

access_start and access_observe return the session and the Except value. Read the latter with lungo_value_is_ok and lungo_value_get_except.

Use a proved function as a test oracle

An oracle is a specification you can run. In the example, evaluate decides a request against a policy, and is proved to decide the relation Permits exactly:

@[lungo_spec "lungo.relation"]
def Permits (policy : List Rule) (r : Request) : Prop :=
  (∃ rule ∈ policy, rule.covers r = true ∧ rule.effect = .allow) ∧
    ∀ rule ∈ policy, rule.covers r = true → rule.effect ≠ .deny

@[lungo_role "lungo.oracle"]
def evaluate (policy : List Rule) (r : Request) : Bool := ...

@[lungo_claim "lungo.decides" subject evaluate spec Permits]
theorem evaluate_decides (policy : List Rule) (r : Request) :
    evaluate policy r = true ↔ Permits policy r := ...

@[lungo_role "lungo.oracle"] records what the export is for. The role appears in the export’s entry in assurance.json and in every language’s view of it, for example __meta::declaration("Access.evaluate").unwrap().assurance.roles in Rust.

Production code that must be fast, such as an evaluator indexed by role and action, is written by hand and tested against the oracle on generated inputs:

for seed in 0..300 {
    let rules = policy(seed, (seed % 8) as usize);
    let index = Index::new(&rules);
    let lean = List::from(rules);
    for role in ROLES {
        for action in ACTIONS {
            let oracle = access::evaluate(lean.clone(), Request { role: role.into(), action: action.into() });
            assert_eq!(index.permits(role, action), oracle, "seed {seed}: {role} {action}");
        }
    }
}

The example’s Python test does the same with a Python evaluator. An evaluator that ignores deny rules fails the test within a few seeds. Because the oracle is proved to decide Permits, agreeing with it means agreeing with the specification on every input tested, not with another implementation that might share the bug.

See also

• How to state claims about exports
• lungo’s Lean library: Lungo.Trace and Lungo.Monitor