sigil/sigil.py

#!/usr/bin/env python3
"""Command line for Sigil.

    sigil.py seed <agent>                     one-time enrolment secret
    sigil.py keygen <path> [pass]               write an Ed25519 key (0600, encrypted if given)
    sigil.py enrol <agent> <pubkey.b64> <secret>
    sigil.py challenge <agent>                nonce to sign
    sigil.py checkpoint <key> <agent> [label] sign the current audit head
                                              (also publishes off-box when SIGIL_ANCHOR is set)
    sigil.py verify-chain                     check chain, checkpoints, and off-box tip
    sigil.py revoke <agent> [--until UNIX] [--reason R] [--actor A]
                                         --until omitted means permanent
    sigil.py reinstate <agent> [actor]       lift a revocation
    sigil.py sweep                            prune stale nonces and rate rows
    sigil.py sign <keypath> <nonce> [pass]    signature, base64
    sigil.py verify <nonce> <sig.b64>         agent session token
    sigil.py code <agent> <keypath> [pass]    one-time code, proving the key
    sigil.py redeem <code>                    human_via_agent session token
    sigil.py whoami <token>                   describe a session
    sigil.py events [actor]                   audit trail

Uses SIGIL_DB (default ./sigil.db), SIGIL_KEY (default ./agent.key),
optional SIGIL_ANCHOR (filesystem path for the off-box signed tip), and
optional SIGIL_SEAL_PASS (when set, SIGIL_DB is an AES-GCM sealed file).
"""

from __future__ import annotations

import base64
import json
import os
import sys

from sigil import errors, identity, keys, sessions, viewcodes
from sigil.sealed import SealedStore, is_sealed_file
from sigil.store import Store

DB = os.environ.get("SIGIL_DB", "./sigil.db")
KEY = os.environ.get("SIGIL_KEY", "./agent.key")
ANCHOR = os.environ.get("SIGIL_ANCHOR")  # off-box tip path; unset = in-DB only
SEAL_PASS = os.environ.get("SIGIL_SEAL_PASS")  # seal passphrase; unset = cleartext DB


def out(obj) -> None:
    print(json.dumps(obj, indent=2, sort_keys=True))


def main(argv) -> int:
    if len(argv) < 2:
        print(__doc__)
        return 2
    cmd = argv[1]
    store = None
    if SEAL_PASS or (os.path.exists(DB) and is_sealed_file(DB)):
        if not SEAL_PASS:
            print('{"error":"ValueError","message":"SIGIL_DB is sealed; set SIGIL_SEAL_PASS"}')
            return 1
        store = SealedStore(DB, SEAL_PASS, anchor_path=ANCHOR)
    else:
        store = Store(DB, anchor_path=ANCHOR)

    try:
        if cmd == "seed":
            agent = argv[2]
            out({"agent": agent, "secret": identity.seed_bootstrap(store, agent)})

        elif cmd == "keygen":
            path = argv[2] if len(argv) > 2 else KEY
            private, public = keys.generate()
            passphrase = argv[3] if len(argv) > 3 else None
            keys.save_private(private, path, passphrase)
            out({
                "private_key": path,
                "fingerprint": keys.fingerprint(public),
                "public_b64": keys.export_public_b64(public),
            })

        elif cmd == "enrol":
            out({"fingerprint": identity.enrol(store, argv[2], argv[3], argv[4])})

        elif cmd == "challenge":
            out(identity.challenge(store, argv[2]))

        elif cmd == "sign":
            private = keys.load_private(argv[2], argv[4] if len(argv) > 4 else None)
            nonce = argv[3]
            out({"nonce": nonce, "signature": base64.b64encode(
                keys.sign(private, nonce.encode("utf-8"))).decode("ascii")})

        elif cmd == "verify":
            out(identity.verify(store, argv[2], argv[3]))

        elif cmd == "code":
            # Mints durable authority, so it proves key possession rather than
            # trusting a name on the command line.
            if len(argv) < 4:
                print("usage: sigil.py code <agent> <keypath> [passphrase]")
                return 2
            private = keys.load_private(argv[3], argv[4] if len(argv) > 4 else None)
            chal = identity.challenge(store, argv[2])
            signed = keys.sign(private, chal["nonce"].encode("utf-8"))
            sig = base64.b64encode(signed).decode("ascii")
            token = identity.verify(store, chal["nonce"], sig)["token"]
            out(viewcodes.issue_code(store, token))

        elif cmd == "redeem":
            out(viewcodes.redeem_code(store, argv[2]))

        elif cmd == "whoami":
            session = sessions.resolve(store, argv[2])
            may_act = True
            try:
                sessions.assert_may_act_as_agent(session)
            except errors.AgentImpersonationDenied:
                may_act = False
            session["may_act_as_agent"] = may_act
            out(session)

        elif cmd == "checkpoint":
            # Signs the chain head with a key the database does not hold, which
            # is what makes a later rewrite detectable.
            if len(argv) < 4:
                print("usage: sigil.py checkpoint <keypath> <agent> [label] [pass]")
                return 2
            private = keys.load_private(argv[2], argv[5] if len(argv) > 5 else None)
            out(store.checkpoint(private, argv[3],
                                 argv[4] if len(argv) > 4 else None))

        elif cmd == "verify-chain":
            chain = store.verify_chain()
            checkpoints = store.verify_checkpoints()
            external = store.verify_external_anchors()
            out({"chain": chain, "checkpoints": checkpoints,
                 "external_anchors": external})
            ok = chain["ok"] and checkpoints["ok"] and external["ok"]
            return 0 if ok else 1

        elif cmd == "revoke":
            # Positional optional args are a trap here: "revoke alice please"
            # would otherwise try to parse "please" as a timestamp. Named flags
            # are unambiguous, and a number only counts as a time if it parses.
            allowed = {"until", "reason", "actor"}
            flags, rest = {}, []
            i = 2
            while i < len(argv):
                if argv[i].startswith("--"):
                    key = argv[i][2:]
                    if key not in allowed:
                        print(f"unknown flag --{key}; expected one of "
                              f"{', '.join('--' + a for a in sorted(allowed))}")
                        return 2
                    flags[key] = argv[i + 1] if i + 1 < len(argv) else ""
                    i += 2
                else:
                    rest.append(argv[i])
                    i += 1
            if len(rest) != 1:
                print("usage: sigil.py revoke <agent> [--until N] [--reason R]"
                      " [--actor A]")
                if rest:
                    # Refusing beats guessing which agent was meant.
                    print(f"expected exactly one agent name, got {rest}")
                return 2
            until = None
            if "until" in flags:
                try:
                    until = int(flags["until"])
                except ValueError:
                    print(f"--until must be a unix timestamp, got {flags['until']!r}")
                    return 2
            killed = identity.revoke(
                store, rest[0], until=until,
                reason=flags.get("reason"),
                actor=flags.get("actor", "system"),
            )
            out({"agent": rest[0], "revoked": True,
                 "permanent": until is None, "sessions_revoked": killed})


        elif cmd == "reinstate":
            identity.reinstate(store, argv[2],
                               argv[3] if len(argv) > 3 else "system")
            out({"agent": argv[2], "revoked": False})

        elif cmd == "sweep":
            out(store.sweep())

        elif cmd == "events":
            out(store.events(argv[2] if len(argv) > 2 else None,
                             int(argv[3]) if len(argv) > 3 else 50))

        else:
            print(__doc__)
            return 2

    except errors.SigilError as exc:
        print(json.dumps({"error": type(exc).__name__, "message": str(exc)}))
        _flush_if_sealed(store)
        return 1
    except (ValueError, OSError) as exc:
        # Wrong passphrase, unreadable key file. Report it as a plain result
        # rather than a traceback: it is an ordinary operator mistake.
        print(json.dumps({"error": type(exc).__name__, "message": str(exc)[:200]}))
        _flush_if_sealed(store)
        return 1

    _flush_if_sealed(store)
    return 0


def _flush_if_sealed(store) -> None:
    if store is not None and isinstance(store, SealedStore):
        try:
            store.flush()
        except Exception:
            pass


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))