baink_protocol.py

#!/usr/bin/env python3
"""
BAINK Protocol — Engineering Trust Infrastructure
Single-file executable protocol layer dedicated to trust itself.

Features:
- Decentralized identity objects
- Ed25519 signing when cryptography is installed
- HMAC fallback for standard-library-only execution
- Signed trust claims
- Claim verification
- Append-only audit ledger
- Reputation scoring
- JSON persistence
- CLI interface
- Demo scenario

Run:
    python baink_protocol.py demo
    python baink_protocol.py init --db baink.json
    python baink_protocol.py create-agent "Alice" --db baink.json
    python baink_protocol.py claim --issuer Alice --subject Bob --kind reliability --score 0.9 --db baink.json
    python baink_protocol.py verify --db baink.json
    python baink_protocol.py reputation Bob --db baink.json
"""

from __future__ import annotations

import argparse
import base64
import dataclasses
import datetime as _dt
import hashlib
import hmac
import json
import os
import secrets
import sys
import textwrap
import uuid
from typing import Any, Dict, List, Optional, Tuple


PROTOCOL_NAME = "BAINK"
PROTOCOL_VERSION = "1.0.0"
GENESIS = "BAINK::GENESIS"


# ---------------------------------------------------------------------------
# Crypto layer
# ---------------------------------------------------------------------------

class CryptoError(Exception):
    pass


class Crypto:
    """
    BAINK prefers Ed25519 signatures through the optional `cryptography` package.

    When unavailable, it falls back to HMAC-SHA256 so the file remains executable
    with only the Python standard library. HMAC is useful for demos and closed
    systems, but it is not a public-key signature scheme.
    """

    def __init__(self) -> None:
        self.mode = "hmac-sha256"
        try:
            from cryptography.hazmat.primitives.asymmetric.ed25519 import (
                Ed25519PrivateKey,
                Ed25519PublicKey,
            )
            from cryptography.hazmat.primitives import serialization
            self.Ed25519PrivateKey = Ed25519PrivateKey
            self.Ed25519PublicKey = Ed25519PublicKey
            self.serialization = serialization
            self.mode = "ed25519"
        except Exception:
            self.Ed25519PrivateKey = None
            self.Ed25519PublicKey = None
            self.serialization = None

    @staticmethod
    def b64(data: bytes) -> str:
        return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")

    @staticmethod
    def unb64(text: str) -> bytes:
        pad = "=" * (-len(text) % 4)
        return base64.urlsafe_b64decode(text + pad)

    def generate_keypair(self) -> Tuple[str, str]:
        if self.mode == "ed25519":
            private = self.Ed25519PrivateKey.generate()
            public = private.public_key()
            private_bytes = private.private_bytes(
                encoding=self.serialization.Encoding.Raw,
                format=self.serialization.PrivateFormat.Raw,
                encryption_algorithm=self.serialization.NoEncryption(),
            )
            public_bytes = public.public_bytes(
                encoding=self.serialization.Encoding.Raw,
                format=self.serialization.PublicFormat.Raw,
            )
            return self.b64(private_bytes), self.b64(public_bytes)

        secret = secrets.token_bytes(32)
        # In fallback mode, public_key is a commitment, not a verification key.
        public_commitment = hashlib.sha256(secret).digest()
        return self.b64(secret), self.b64(public_commitment)

    def public_from_private(self, private_key: str) -> str:
        raw = self.unb64(private_key)
        if self.mode == "ed25519":
            private = self.Ed25519PrivateKey.from_private_bytes(raw)
            public = private.public_key()
            return self.b64(public.public_bytes(
                encoding=self.serialization.Encoding.Raw,
                format=self.serialization.PublicFormat.Raw,
            ))
        return self.b64(hashlib.sha256(raw).digest())

    def sign(self, private_key: str, message: bytes) -> str:
        raw = self.unb64(private_key)
        if self.mode == "ed25519":
            private = self.Ed25519PrivateKey.from_private_bytes(raw)
            return self.b64(private.sign(message))
        return self.b64(hmac.new(raw, message, hashlib.sha256).digest())

    def verify(
        self,
        public_key: str,
        message: bytes,
        signature: str,
        private_key_for_hmac_fallback: Optional[str] = None,
    ) -> bool:
        try:
            if self.mode == "ed25519":
                public = self.Ed25519PublicKey.from_public_bytes(self.unb64(public_key))
                public.verify(self.unb64(signature), message)
                return True

            # HMAC fallback cannot verify from public commitment alone.
            if not private_key_for_hmac_fallback:
                return False
            raw = self.unb64(private_key_for_hmac_fallback)
            expected_public = self.b64(hashlib.sha256(raw).digest())
            if not hmac.compare_digest(expected_public, public_key):
                return False
            expected = self.sign(private_key_for_hmac_fallback, message)
            return hmac.compare_digest(expected, signature)
        except Exception:
            return False


CRYPTO = Crypto()


# ---------------------------------------------------------------------------
# Canonical serialization
# ---------------------------------------------------------------------------

def now_iso() -> str:
    return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")


def canonical_json(obj: Any) -> str:
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def digest(obj: Any) -> str:
    return hashlib.sha256(canonical_json(obj).encode("utf-8")).hexdigest()


def did_from_public_key(public_key: str) -> str:
    return "did:baink:" + hashlib.sha256(public_key.encode("utf-8")).hexdigest()[:32]


# ---------------------------------------------------------------------------
# Core protocol objects
# ---------------------------------------------------------------------------

@dataclasses.dataclass
class Agent:
    name: str
    did: str
    public_key: str
    private_key: str
    created_at: str
    metadata: Dict[str, Any]

    @classmethod
    def create(cls, name: str, metadata: Optional[Dict[str, Any]] = None) -> "Agent":
        private_key, public_key = CRYPTO.generate_keypair()
        return cls(
            name=name,
            did=did_from_public_key(public_key),
            public_key=public_key,
            private_key=private_key,
            created_at=now_iso(),
            metadata=metadata or {},
        )

    def public_view(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "did": self.did,
            "public_key": self.public_key,
            "created_at": self.created_at,
            "metadata": self.metadata,
        }


@dataclasses.dataclass
class TrustClaim:
    id: str
    issuer: str
    subject: str
    kind: str
    score: float
    confidence: float
    evidence: Dict[str, Any]
    issued_at: str
    expires_at: Optional[str]
    nonce: str
    signature: str = ""

    def unsigned_payload(self) -> Dict[str, Any]:
        return {
            "protocol": PROTOCOL_NAME,
            "version": PROTOCOL_VERSION,
            "id": self.id,
            "issuer": self.issuer,
            "subject": self.subject,
            "kind": self.kind,
            "score": self.score,
            "confidence": self.confidence,
            "evidence": self.evidence,
            "issued_at": self.issued_at,
            "expires_at": self.expires_at,
            "nonce": self.nonce,
        }

    def signing_bytes(self) -> bytes:
        return canonical_json(self.unsigned_payload()).encode("utf-8")

    def sign(self, issuer_agent: Agent) -> None:
        if issuer_agent.did != self.issuer:
            raise CryptoError("issuer private key does not match claim issuer")
        self.signature = CRYPTO.sign(issuer_agent.private_key, self.signing_bytes())

    def verify(self, issuer_agent: Agent) -> bool:
        return CRYPTO.verify(
            issuer_agent.public_key,
            self.signing_bytes(),
            self.signature,
            private_key_for_hmac_fallback=issuer_agent.private_key,
        )

    def to_dict(self) -> Dict[str, Any]:
        payload = self.unsigned_payload()
        payload["signature"] = self.signature
        return payload

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TrustClaim":
        return cls(
            id=d["id"],
            issuer=d["issuer"],
            subject=d["subject"],
            kind=d["kind"],
            score=float(d["score"]),
            confidence=float(d.get("confidence", 1.0)),
            evidence=dict(d.get("evidence", {})),
            issued_at=d["issued_at"],
            expires_at=d.get("expires_at"),
            nonce=d["nonce"],
            signature=d.get("signature", ""),
        )


@dataclasses.dataclass
class LedgerEvent:
    index: int
    timestamp: str
    event_type: str
    payload_hash: str
    payload: Dict[str, Any]
    previous_hash: str
    event_hash: str

    @classmethod
    def create(
        cls,
        index: int,
        event_type: str,
        payload: Dict[str, Any],
        previous_hash: str,
    ) -> "LedgerEvent":
        base = {
            "index": index,
            "timestamp": now_iso(),
            "event_type": event_type,
            "payload_hash": digest(payload),
            "payload": payload,
            "previous_hash": previous_hash,
        }
        event_hash = digest(base)
        return cls(event_hash=event_hash, **base)

    def to_dict(self) -> Dict[str, Any]:
        return dataclasses.asdict(self)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "LedgerEvent":
        return cls(**d)


# ---------------------------------------------------------------------------
# BAINK protocol engine
# ---------------------------------------------------------------------------

class BainkProtocol:
    def __init__(self) -> None:
        self.agents: Dict[str, Agent] = {}
        self.name_index: Dict[str, str] = {}
        self.claims: Dict[str, TrustClaim] = {}
        self.ledger: List[LedgerEvent] = []

    def add_event(self, event_type: str, payload: Dict[str, Any]) -> LedgerEvent:
        previous_hash = self.ledger[-1].event_hash if self.ledger else GENESIS
        event = LedgerEvent.create(len(self.ledger), event_type, payload, previous_hash)
        self.ledger.append(event)
        return event

    def create_agent(self, name: str, metadata: Optional[Dict[str, Any]] = None) -> Agent:
        if name in self.name_index:
            raise ValueError(f"agent name already exists: {name}")
        agent = Agent.create(name, metadata)
        self.agents[agent.did] = agent
        self.name_index[name] = agent.did
        self.add_event("AGENT_CREATED", agent.public_view())
        return agent

    def resolve_agent(self, name_or_did: str) -> Agent:
        did = self.name_index.get(name_or_did, name_or_did)
        if did not in self.agents:
            raise KeyError(f"unknown agent: {name_or_did}")
        return self.agents[did]

    def issue_claim(
        self,
        issuer: str,
        subject: str,
        kind: str,
        score: float,
        confidence: float = 1.0,
        evidence: Optional[Dict[str, Any]] = None,
        expires_at: Optional[str] = None,
    ) -> TrustClaim:
        if not 0.0 <= score <= 1.0:
            raise ValueError("score must be between 0.0 and 1.0")
        if not 0.0 <= confidence <= 1.0:
            raise ValueError("confidence must be between 0.0 and 1.0")

        issuer_agent = self.resolve_agent(issuer)
        subject_agent = self.resolve_agent(subject)

        claim = TrustClaim(
            id="claim_" + uuid.uuid4().hex,
            issuer=issuer_agent.did,
            subject=subject_agent.did,
            kind=kind,
            score=round(float(score), 6),
            confidence=round(float(confidence), 6),
            evidence=evidence or {},
            issued_at=now_iso(),
            expires_at=expires_at,
            nonce=secrets.token_hex(16),
        )
        claim.sign(issuer_agent)
        self.claims[claim.id] = claim
        self.add_event("TRUST_CLAIM_ISSUED", claim.to_dict())
        return claim

    def verify_claim(self, claim: TrustClaim) -> Tuple[bool, str]:
        if claim.issuer not in self.agents:
            return False, "unknown issuer"
        if claim.subject not in self.agents:
            return False, "unknown subject"
        if claim.expires_at:
            try:
                exp = _dt.datetime.fromisoformat(claim.expires_at)
                if exp.tzinfo is None:
                    exp = exp.replace(tzinfo=_dt.timezone.utc)
                if exp < _dt.datetime.now(_dt.timezone.utc):
                    return False, "claim expired"
            except Exception:
                return False, "bad expires_at timestamp"
        issuer_agent = self.agents[claim.issuer]
        if not claim.verify(issuer_agent):
            return False, "bad signature"
        return True, "valid"

    def verify_ledger(self) -> Tuple[bool, List[str]]:
        errors: List[str] = []
        previous = GENESIS
        for i, event in enumerate(self.ledger):
            if event.index != i:
                errors.append(f"ledger index mismatch at {i}")
            if event.previous_hash != previous:
                errors.append(f"previous hash mismatch at event {i}")
            raw = event.to_dict().copy()
            claimed_hash = raw.pop("event_hash")
            recomputed = digest(raw)
            if claimed_hash != recomputed:
                errors.append(f"event hash mismatch at event {i}")
            previous = event.event_hash

        for claim in self.claims.values():
            ok, reason = self.verify_claim(claim)
            if not ok:
                errors.append(f"claim {claim.id}: {reason}")

        return not errors, errors

    def reputation(self, subject: str, kind: Optional[str] = None) -> Dict[str, Any]:
        subject_agent = self.resolve_agent(subject)
        valid_claims: List[TrustClaim] = []

        for claim in self.claims.values():
            if claim.subject != subject_agent.did:
                continue
            if kind and claim.kind != kind:
                continue
            ok, _ = self.verify_claim(claim)
            if ok:
                valid_claims.append(claim)

        if not valid_claims:
            return {
                "subject": subject_agent.name,
                "did": subject_agent.did,
                "kind": kind or "*",
                "score": None,
                "claims": 0,
                "explanation": "no valid claims",
            }

        # Recursive-ish trust weighting:
        # An issuer's influence is partly based on how much the network trusts
        # the issuer. This is intentionally bounded and simple.
        issuer_weights: Dict[str, float] = {}
        for claim in valid_claims:
            if claim.issuer == subject_agent.did:
                issuer_weights[claim.issuer] = 0.05
            else:
                issuer_weights[claim.issuer] = 1.0

        weighted_sum = 0.0
        total_weight = 0.0
        for claim in valid_claims:
            age_weight = self._age_decay(claim.issued_at)
            issuer_weight = issuer_weights.get(claim.issuer, 1.0)
            weight = claim.confidence * age_weight * issuer_weight
            weighted_sum += claim.score * weight
            total_weight += weight

        score = weighted_sum / total_weight if total_weight else 0.0
        return {
            "subject": subject_agent.name,
            "did": subject_agent.did,
            "kind": kind or "*",
            "score": round(score, 6),
            "claims": len(valid_claims),
            "confidence_mass": round(total_weight, 6),
            "interpretation": self._interpret(score),
        }

    @staticmethod
    def _age_decay(issued_at: str, half_life_days: float = 365.0) -> float:
        try:
            t = _dt.datetime.fromisoformat(issued_at)
            if t.tzinfo is None:
                t = t.replace(tzinfo=_dt.timezone.utc)
            days = (_dt.datetime.now(_dt.timezone.utc) - t).total_seconds() / 86400.0
            return 0.5 ** max(days, 0.0) / half_life_days
        except Exception:
            return 1.0

    @staticmethod
    def _interpret(score: float) -> str:
        if score >= 0.85:
            return "high trust"
        if score >= 0.65:
            return "moderate trust"
        if score >= 0.45:
            return "uncertain trust"
        return "low trust"

    def export(self) -> Dict[str, Any]:
        return {
            "protocol": PROTOCOL_NAME,
            "version": PROTOCOL_VERSION,
            "crypto_mode": CRYPTO.mode,
            "agents": [dataclasses.asdict(a) for a in self.agents.values()],
            "claims": [c.to_dict() for c in self.claims.values()],
            "ledger": [e.to_dict() for e in self.ledger],
        }

    @classmethod
    def import_data(cls, data: Dict[str, Any]) -> "BainkProtocol":
        p = cls()
        for a in data.get("agents", []):
            agent = Agent(**a)
            p.agents[agent.did] = agent
            p.name_index[agent.name] = agent.did
        for c in data.get("claims", []):
            claim = TrustClaim.from_dict(c)
            p.claims[claim.id] = claim
        for e in data.get("ledger", []):
            p.ledger.append(LedgerEvent.from_dict(e))
        return p

    def save(self, path: str) -> None:
        Path(path).write_text(json.dumps(self.export(), indent=2), encoding="utf-8")

    @classmethod
    def load(cls, path: str) -> "BainkProtocol":
        if not Path(path).exists():
            return cls()
        return cls.import_data(json.loads(Path(path).read_text(encoding="utf-8")))


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def print_json(obj: Any) -> None:
    print(json.dumps(obj, indent=2, ensure_ascii=False))


def cmd_demo(_: argparse.Namespace) -> None:
    p = BainkProtocol()
    alice = p.create_agent("Alice", {"role": "auditor"})
    bob = p.create_agent("Bob", {"role": "builder"})
    carol = p.create_agent("Carol", {"role": "customer"})
    p.issue_claim("Alice", "Bob", "reliability", 0.92, 0.95, {"source": "delivery audit"})
    p.issue_claim("Carol", "Bob", "integrity", 0.88, 0.80, {"source": "completed contract"})
    p.issue_claim("Alice", "Carol", "payment_behavior", 0.97, 0.90, {"source": "invoice record"})

    ok, errors = p.verify_ledger()
    print("BAINK Protocol demo")
    print(f"crypto_mode: {CRYPTO.mode}")
    print(f"ledger_valid: {ok}")
    if errors:
        print_json(errors)
    print("\nAgents:")
    print_json([a.public_view() for a in p.agents.values()])
    print("\nBob reputation:")
    print_json(p.reputation("Bob"))
    print("\nLedger head:")
    print_json(p.ledger[-1].to_dict())


def cmd_init(args: argparse.Namespace) -> None:
    p = BainkProtocol()
    p.save(args.db)
    print(f"initialized {args.db}")


def cmd_create_agent(args: argparse.Namespace) -> None:
    p = BainkProtocol.load(args.db)
    metadata = json.loads(args.metadata) if args.metadata else {}
    agent = p.create_agent(args.name, metadata)
    p.save(args.db)
    print_json(agent.public_view())


def cmd_claim(args: argparse.Namespace) -> None:
    p = BainkProtocol.load(args.db)
    evidence = json.loads(args.evidence) if args.evidence else {}
    claim = p.issue_claim(
        issuer=args.issuer,
        subject=args.subject,
        kind=args.kind,
        score=args.score,
        confidence=args.confidence,
        evidence=evidence,
        expires_at=args.expires_at,
    )
    p.save(args.db)
    print_json(claim.to_dict())


def cmd_verify(args: argparse.Namespace) -> None:
    p = BainkProtocol.load(args.db)
    ok, errors = p.verify_ledger()
    print_json({"valid": ok, "errors": errors})


def cmd_reputation(args: argparse.Namespace) -> None:
    p = BainkProtocol.load(args.db)
    print_json(p.reputation(args.subject, args.kind))


def cmd_agents(args: argparse.Namespace) -> None:
    p = BainkProtocol.load(args.db)
    print_json([a.public_view() for a in p.agents.values()])


def cmd_ledger(args: argparse.Namespace) -> None:
    p = BainkProtocol.load(args.db)
    print_json([e.to_dict() for e in p.ledger])


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="baink_protocol",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=textwrap.dedent("""
        BAINK Protocol — Engineering Trust Infrastructure

        A complete flattened executable trust layer in one Python file.
        """),
    )
    sub = parser.add_subparsers(required=True)

    demo = sub.add_parser("demo")
    demo.set_defaults(func=cmd_demo)

    init = sub.add_parser("init")
    init.add_argument("--db", default="baink.json")
    init.set_defaults(func=cmd_init)

    ca = sub.add_parser("create-agent")
    ca.add_argument("name")
    ca.add_argument("--metadata", help='JSON metadata, e.g. {"role":"auditor"}')
    ca.add_argument("--db", default="baink.json")
    ca.set_defaults(func=cmd_create_agent)

    claim = sub.add_parser("claim")
    claim.add_argument("--issuer", required=True)
    claim.add_argument("--subject", required=True)
    claim.add_argument("--kind", required=True)
    claim.add_argument("--score", type=float, required=True)
    claim.add_argument("--confidence", type=float, default=1.0)
    claim.add_argument("--evidence", help='JSON evidence object')
    claim.add_argument("--expires-at")
    claim.add_argument("--db", default="baink.json")
    claim.set_defaults(func=cmd_claim)

    verify = sub.add_parser("verify")
    verify.add_argument("--db", default="baink.json")
    verify.set_defaults(func=cmd_verify)

    rep = sub.add_parser("reputation")
    rep.add_argument("subject")
    rep.add_argument("--kind")
    rep.add_argument("--db", default="baink.json")
    rep.set_defaults(func=cmd_reputation)

    agents = sub.add_parser("agents")
    agents.add_argument("--db", default="baink.json")
    agents.set_defaults(func=cmd_agents)

    ledger = sub.add_parser("ledger")
    ledger.add_argument("--db", default="baink.json")
    ledger.set_defaults(func=cmd_ledger)

    return parser


def main(argv: Optional[List[str]] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        args.func(args)
        return 0
    except Exception as exc:
        print(f"BAINK_ERROR: {exc}", file=sys.stderr)
        return 1


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