Replay a historical policy decision

python -m pip install https://docs.soterias.io/downloads/api-preview/soterias_local-0.1.0rc5%2Bapi2-cp311-abi3-win_amd64.whl

npm install https://docs.soterias.io/downloads/api-preview/soterias-core-0.1.0-rc.5-api.2.tgz
npm install --save-dev typescript@5.8.3 @types/node@24

import json
import os
from pathlib import Path
from soterias import audit, PolicyReplayError

import assert from 'node:assert/strict';
import fs from 'node:fs';
import path from 'node:path';
import { audit, HistoricalPolicyInputs } from '@soterias/core';

# %% Load historical data and independent trust
# The operator supplies the input pin; do not derive it from an untrusted capture.
root = Path(os.environ["SOTERIAS_REPLAY"])
captured = audit.PolicyInputs((root / "allow-input.json").read_bytes())
pin = (root / "allow-pin.txt").read_text(encoding="utf-8").strip()
anchor = (root / "allow-anchor.json").read_bytes()

// %% Load historical data and independent trust
// The operator supplies the input pin; do not derive it from an untrusted capture.
const root = process.env.SOTERIAS_REPLAY!;
const captured = new HistoricalPolicyInputs(fs.readFileSync(path.join(root, 'allow-input.json')));
const pin = fs.readFileSync(path.join(root, 'allow-pin.txt'), 'utf8').trim();
const anchor = fs.readFileSync(path.join(root, 'allow-anchor.json'));

# %% Evaluate without dispatching the historical effect
# Rust uses the captured historical policy and evaluation time.
result = audit.replay(captured, expected_input_digest=pin,
                      expected_trust_anchor=anchor)
data = result.data()
assert data["scope"] == "historical_policy_only"

// %% Evaluate without dispatching the historical effect
// Rust uses the captured historical policy and evaluation time.
const result = audit.replay(captured, { expectedInputDigest: pin, expectedTrustAnchor: anchor });
const data = result.data();
assert.equal(data.scope, 'historical_policy_only');

# %% Refuse a changed independent pin
# A replay result does not authorize a current operation.
try:
    audit.replay(captured, expected_input_digest="0" * 64,
                 expected_trust_anchor=anchor)
except PolicyReplayError:
    pass
else:
    raise AssertionError("A changed pin was accepted")
print(json.dumps({"status": "PASS", "decision": data["evaluation"]["decision"],
                  "scope": data["scope"]}))

// %% Refuse a changed independent pin
// A replay result does not authorize a current operation.
assert.throws(() => audit.replay(captured, { expectedInputDigest: '0'.repeat(64), expectedTrustAnchor: anchor }));
console.log(JSON.stringify({ status: 'PASS', decision: data.evaluation.decision, scope: data.scope }));

import { Soterias, observe, canonicalize, Metrics, audit, HistoricalPolicyInputs } from '@soterias/core';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { McpServer } from '@modelcontextprotocol/server';
import { serveStdio } from '@modelcontextprotocol/server/stdio';
import * as z from 'zod/v4';

// Trusted startup: paths, trust and credentials are fixed by the host.
const read = name => fs.readFileSync(path.join(process.env.SOTERIAS_REPLAY, name));
const capture = new HistoricalPolicyInputs(read('allow-input.json'));
const options = {expectedInputDigest: read('allow-pin.txt').toString().trim(), expectedTrustAnchor: read('allow-anchor.json')};

// Agent input is bounded data. It cannot replace host configuration or trust.
const handle = serveStdio(() => {
  const server = new McpServer({name: 'soterias-replay', version: '1.0.0'});
  server.registerTool('replay_policy', {
    description: 'Use replay through the host-selected native SDK.',
    inputSchema: z.object({}).strict(),
  }, async input => {
    try {
      const result = await (async () => {
        return audit.replay(capture, options).data();
      })();

      return {content: [{type: 'text', text: JSON.stringify(result, (_key, value) => typeof value === 'bigint' ? value.toString() : value)}]};
    } catch {
      return {isError: true, content: [{type: 'text', text: 'feature_request_refused'}]};
    }
  });
  return server;
}, {maxSubscriptions: 0});
process.stdin.once('end', () => {
  void (async () => { await handle.close();  })();
});

import { Client } from '@modelcontextprotocol/client';
import { StdioClientTransport } from '@modelcontextprotocol/client/stdio';
import { fileURLToPath } from 'node:url';

const client = new Client({name: 'feature-example', version: '1.0.0'});
try {
  await client.connect(new StdioClientTransport({
    command: process.execPath,
    args: [fileURLToPath(new URL('./mcp.mjs', import.meta.url))],
    env: Object.fromEntries(Object.entries(process.env).filter(([, value]) => value !== undefined)),
  }));
  const result = await client.callTool({name: 'replay_policy', arguments: {}});
  if (result.isError) throw new Error('Feature request refused');
  console.log(JSON.stringify(result));
} finally { await client.close(); }