Execute a protected SQLite operation

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
import tempfile
import time
from soterias import security


def save_artifact(name, data):
    # The operator supplies an existing private parent; create files without replacement.
    descriptor = os.open(output / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(data)
        stream.flush()
        os.fsync(stream.fileno())

import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import readline from 'node:readline/promises';
import { security, type ExecutorConfig } from '@soterias/core';

# %% Load operator-selected inputs
# These paths and the unlock come from the trusted application, never an agent.
root = Path(os.environ["SOTERIAS_ROOT"])
custody = Path(os.environ["SOTERIAS_CUSTODY"])
checkpoint = Path(os.environ["SOTERIAS_CHECKPOINT"]).read_bytes()
unlock = os.environ["SOTERIAS_UNLOCK"].encode()
configuration = (root / "document-profile.json").read_bytes()
request = (root / "allow-request.json").read_bytes()


def config(pin):
    return security.Config(root, custody=custody, configuration=configuration,
                           checkpoint=pin, passphrase=unlock)

// %% Load operator-selected inputs
// These paths and the unlock come from the trusted application, never an agent.
function required(name: string) {
  const value = process.env[name];
  if (!value) throw new Error(`Set ${name} from your trusted host configuration`);
  return value;
}
const root = required('SOTERIAS_ROOT'), custody = required('SOTERIAS_CUSTODY');
const checkpoint = fs.readFileSync(required('SOTERIAS_CHECKPOINT'));
const passphrase = Buffer.from(required('SOTERIAS_UNLOCK'));
const configuration = fs.readFileSync(path.join(root, 'document-profile.json'));
const request = fs.readFileSync(path.join(root, 'allow-request.json'));
const config = (pin: Buffer): ExecutorConfig => ({ root, custody, configuration,
  checkpoint: pin, passphrase, profile: 'documents' });

# %% Prepare an exact operation without executing it
# Preparing produces the native binding and retained pin used by the reviewer.
with security.Executor(config(checkpoint)) as executor:
    operation = executor.prepare(request)

// %% Prepare an exact operation without executing it
// Preparing produces the native binding and retained pin used by the reviewer.
let executor = new security.Executor(config(checkpoint));
let operation;
try { operation = executor.prepare(request); } finally { executor.close(); }

# %% Ask the separate reviewer to inspect the original selection
# This is trusted reviewer UI code. Never expose this owner as an agent tool.
with security.Reviewer(config(operation.checkpoint),
        signing_key=json.loads((root / "operation-key.json").read_bytes())) as reviewer:
    review = reviewer.prepare(request, operation,
                              expires_at_ms=int(time.time() * 1000) + 300_000)
    print(review.selection.decode())
    answer = input("Inspect this selection, then type APPROVE: ")
    print()
    if answer != "APPROVE":
        raise RuntimeError("Review declined")
    approval = reviewer.approve(review, confirmation=review.confirmation_digest)

// %% Ask the separate reviewer to inspect the original selection
// This is trusted reviewer UI code. Never expose this owner as an agent tool.
const reviewer = new security.Reviewer({ ...config(operation.checkpoint),
  profile: 'documents', signingKey: JSON.parse(fs.readFileSync(path.join(root, 'operation-key.json'), 'utf8')) });
const terminal = readline.createInterface({ input: process.stdin, output: process.stdout });
let approval;
try {
  const review = reviewer.prepare(request, operation, { expiresAtMs: BigInt(Date.now()) + 300000n });
  console.log(review.selection.toString());
  const answer = await terminal.question('Inspect this selection, then type APPROVE: ');
  console.log();
  if (answer !== 'APPROVE')
    throw new Error('Review declined');
  approval = reviewer.approve(review, { confirmation: review.confirmationDigest });
} finally { terminal.close(); reviewer.close(); }

# %% Execute through Rust and retain the next checkpoint
# A signed approval still passes through native policy, warrant and broker checks.
output = Path(os.environ["SOTERIAS_OUTPUT"])
with security.Executor(config(operation.checkpoint)) as executor:
    report = executor.invoke(request, approval=approval)
    save_artifact("checkpoint.json", executor.checkpoint())
    for index, receipt in enumerate(report.receipts):
        save_artifact(f"receipt-{index}.json", receipt)
    print(json.dumps({"status": report.status, "receipt_count": len(report.receipts),
                      "target_observed": report.target_observed,
                      "native_timing": json.loads(report.native_timing),
                      "receipt_errors": report.receipt_errors,
                      "evidence_directory": str(output)}))

// %% Execute through Rust and retain the next checkpoint
// A signed approval still passes through native policy, warrant and broker checks.
const output = required('SOTERIAS_OUTPUT');
executor = new security.Executor(config(operation.checkpoint));
try {
  const report = await executor.invokeAsync(request, { approval });
  fs.writeFileSync(path.join(output, 'checkpoint.json'), executor.checkpoint(), { flag: 'wx', mode: 0o600 });
  report.receipts.forEach((receipt, index) => fs.writeFileSync(path.join(output, `receipt-${index}.json`), receipt));
  console.log(JSON.stringify({ status: report.status, receipt_count: report.receipts.length,
    target_observed: report.targetObserved, native_timing: JSON.parse(report.nativeTiming.toString()),
    receipt_errors: report.receiptErrors, evidence_directory: output }));
} finally { await executor.closeAsync(); }

import fs from 'node:fs';
import path from 'node:path';
import { security } from '@soterias/core';
import { listenAccountingHttp } from '@soterias/mcp-server/accounting-http';

// %% Open the operator-selected native owner
// Enrollment, tool review, warrants and exact operation approval happen beforehand.
const root = process.env.SOTERIAS_ROOT;
const read = name => fs.readFileSync(path.join(root, name));
const owner = security.Host.open({ root, custody: process.env.SOTERIAS_CUSTODY,
  configuration: read('configuration.json'),
  checkpoint: fs.readFileSync(process.env.SOTERIAS_CHECKPOINT),
  passphrase: Buffer.from(process.env.SOTERIAS_UNLOCK) });

// %% Start a loopback host with authenticated native admission
// The agent cannot replace the policy, custody or signed approval through a tool call.
let listener;
try {
  listener = await listenAccountingHttp(owner, {
    transportPolicy: read('transport-policy.json'),
    operationApproval: read('operation-approval.json'),
  });
  const token = owner.issueTransportCredential().toString('base64url');
  fs.writeFileSync(process.env.SOTERIAS_CONNECTION,
    JSON.stringify({url: listener.url, token}), {flag: 'wx', mode: 0o600});
} catch (error) { owner.close(); throw error; }

// %% Retain the acknowledged checkpoint when shutting down
// Keep this output independently of the writable host and signing custody.
let stopping;
function stop() {
  stopping ??= (async () => {
    await listener.close();
    fs.writeFileSync(process.env.SOTERIAS_NEXT_CHECKPOINT, listener.checkpoint(),
      {flag: 'wx', mode: 0o600});
  })();
  return stopping;
}
for (const signal of ['SIGINT', 'SIGTERM']) process.once(signal, () => {
  void stop().catch(() => { process.stderr.write('host_shutdown_failed\n'); process.exitCode = 1; });
});
// A parent process can close this control stream for orderly local shutdown.
process.stdin.resume();
process.stdin.once('end', () => { void stop(); });

import fs from 'node:fs';
import { Client, StreamableHTTPClientTransport } from '@modelcontextprotocol/client';

// %% Connect to an operator-configured protected MCP host
// The token is scoped to execution. Reviewer keys never enter this client.
const client = new Client({ name: 'invoice-assistant', version: '1.0.0' });
const transport = new StreamableHTTPClientTransport(new URL(process.env.SOTERIAS_MCP_URL),
  { requestInit: { headers: { Authorization: 'Bearer ' + process.env.SOTERIAS_MCP_TOKEN },
    redirect: 'error', credentials: 'omit' } });
try {
  await client.connect(transport);

  // %% Execute through the native host's real accounting tool
  // The host selects routes and approval evidence; the tool receives a request.
  const requestJson = fs.readFileSync(process.env.SOTERIAS_REQUEST, 'utf8');
  const result = await client.callTool({ name: 'execute_accounting', arguments: { requestJson } });
  if (result.isError) throw new Error('Protected tool refused the request');
  const content = result.content.find(block => block.type === 'text');
  if (!content || content.type !== 'text') throw new Error('Expected an accounting outcome');
  const outcome = JSON.parse(content.text);

  // %% Inspect the outcome instead of treating transport success as business success
  // Verify original receipts separately with your own retained trust.
  console.log(JSON.stringify({ status: outcome.status, target_observed: outcome.targetObserved }));
} finally { await client.close(); }