BAINK Protocol

1#!/usr/bin/env python3
2"""
3BAINK Protocol — Engineering Trust Infrastructure
4Single-file executable protocol layer dedicated to trust itself.
5 
6Features:
7- Decentralized identity objects
8- Ed25519 signing when cryptography is installed
9- HMAC fallback for standard-library-only execution
10- Signed trust claims
11- Claim verification
12- Append-only audit ledger
13- Reputation scoring
14- JSON persistence
15- CLI interface
16- Demo scenario
17 
18Run:
19    python baink_protocol.py demo
20    python baink_protocol.py init --db baink.json
21    python baink_protocol.py create-agent "Alice" --db baink.json
22    python baink_protocol.py claim --issuer Alice --subject Bob --kind reliability --score 0.9 --db baink.json
23    python baink_protocol.py verify --db baink.json
24    python baink_protocol.py reputation Bob --db baink.json
25"""
26 
27from __future__ import annotations
28 
29import argparse
30import base64
31import dataclasses
32import datetime as _dt
33import hashlib
34import hmac
35import json
36import os
37import secrets
38import sys
39import textwrap
40import uuid
41from typing import Any, Dict, List, Optional, Tuple
42 
43 
44PROTOCOL_NAME = "BAINK"
45PROTOCOL_VERSION = "1.0.0"
46GENESIS = "BAINK::GENESIS"
47 
48 
49# ---------------------------------------------------------------------------
50# Crypto layer
51# ---------------------------------------------------------------------------
52 
53class CryptoError(Exception):
54    pass
55 
56 
57class Crypto:
58    """
59    BAINK prefers Ed25519 signatures through the optional `cryptography` package.
60 
61    When unavailable, it falls back to HMAC-SHA256 so the file remains executable
62    with only the Python standard library. HMAC is useful for demos and closed
63    systems, but it is not a public-key signature scheme.
64    """
65 
66    def __init__(self) -> None:
67        self.mode = "hmac-sha256"
68        try:
69            from cryptography.hazmat.primitives.asymmetric.ed25519 import (
70                Ed25519PrivateKey,
71                Ed25519PublicKey,
72            )
73            from cryptography.hazmat.primitives import serialization
74            self.Ed25519PrivateKey = Ed25519PrivateKey
75            self.Ed25519PublicKey = Ed25519PublicKey
76            self.serialization = serialization
77            self.mode = "ed25519"
78        except Exception:
79            self.Ed25519PrivateKey = None
80            self.Ed25519PublicKey = None
81            self.serialization = None
82 
83    @staticmethod
84    def b64(data: bytes) -> str:
85        return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
86 
87    @staticmethod
88    def unb64(text: str) -> bytes:
89        pad = "=" * (-len(text) % 4)
90        return base64.urlsafe_b64decode(text + pad)
91 
92    def generate_keypair(self) -> Tuple[str, str]:
93        if self.mode == "ed25519":
94            private = self.Ed25519PrivateKey.generate()
95            public = private.public_key()
96            private_bytes = private.private_bytes(
97                encoding=self.serialization.Encoding.Raw,
98                format=self.serialization.PrivateFormat.Raw,
99                encryption_algorithm=self.serialization.NoEncryption(),
100            )
101            public_bytes = public.public_bytes(
102                encoding=self.serialization.Encoding.Raw,
103                format=self.serialization.PublicFormat.Raw,
104            )
105            return self.b64(private_bytes), self.b64(public_bytes)
106 
107        secret = secrets.token_bytes(32)
108        # In fallback mode, public_key is a commitment, not a verification key.
109        public_commitment = hashlib.sha256(secret).digest()
110        return self.b64(secret), self.b64(public_commitment)
111 
112    def public_from_private(self, private_key: str) -> str:
113        raw = self.unb64(private_key)
114        if self.mode == "ed25519":
115            private = self.Ed25519PrivateKey.from_private_bytes(raw)
116            public = private.public_key()
117            return self.b64(public.public_bytes(
118                encoding=self.serialization.Encoding.Raw,
119                format=self.serialization.PublicFormat.Raw,
120            ))
121        return self.b64(hashlib.sha256(raw).digest())
122 
123    def sign(self, private_key: str, message: bytes) -> str:
124        raw = self.unb64(private_key)
125        if self.mode == "ed25519":
126            private = self.Ed25519PrivateKey.from_private_bytes(raw)
127            return self.b64(private.sign(message))
128        return self.b64(hmac.new(raw, message, hashlib.sha256).digest())
129 
130    def verify(
131        self,
132        public_key: str,
133        message: bytes,
134        signature: str,
135        private_key_for_hmac_fallback: Optional[str] = None,
136    ) -> bool:
137        try:
138            if self.mode == "ed25519":
139                public = self.Ed25519PublicKey.from_public_bytes(self.unb64(public_key))
140                public.verify(self.unb64(signature), message)
141                return True
142 
143            # HMAC fallback cannot verify from public commitment alone.
144            if not private_key_for_hmac_fallback:
145                return False
146            raw = self.unb64(private_key_for_hmac_fallback)
147            expected_public = self.b64(hashlib.sha256(raw).digest())
148            if not hmac.compare_digest(expected_public, public_key):
149                return False
150            expected = self.sign(private_key_for_hmac_fallback, message)
151            return hmac.compare_digest(expected, signature)
152        except Exception:
153            return False
154 
155 
156CRYPTO = Crypto()
157 
158 
159# ---------------------------------------------------------------------------
160# Canonical serialization
161# ---------------------------------------------------------------------------
162 
163def now_iso() -> str:
164    return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")
165 
166 
167def canonical_json(obj: Any) -> str:
168    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
169 
170 
171def digest(obj: Any) -> str:
172    return hashlib.sha256(canonical_json(obj).encode("utf-8")).hexdigest()
173 
174 
175def did_from_public_key(public_key: str) -> str:
176    return "did:baink:" + hashlib.sha256(public_key.encode("utf-8")).hexdigest()[:32]
177 
178 
179# ---------------------------------------------------------------------------
180# Core protocol objects
181# ---------------------------------------------------------------------------
182 
183@dataclasses.dataclass
184class Agent:
185    name: str
186    did: str
187    public_key: str
188    private_key: str
189    created_at: str
190    metadata: Dict[str, Any]
191 
192    @classmethod
193    def create(cls, name: str, metadata: Optional[Dict[str, Any]] = None) -> "Agent":
194        private_key, public_key = CRYPTO.generate_keypair()
195        return cls(
196            name=name,
197            did=did_from_public_key(public_key),
198            public_key=public_key,
199            private_key=private_key,
200            created_at=now_iso(),
201            metadata=metadata or {},
202        )
203 
204    def public_view(self) -> Dict[str, Any]:
205        return {
206            "name": self.name,
207            "did": self.did,
208            "public_key": self.public_key,
209            "created_at": self.created_at,
210            "metadata": self.metadata,
211        }
212 
213 
214@dataclasses.dataclass
215class TrustClaim:
216    id: str
217    issuer: str
218    subject: str
219    kind: str
220    score: float
221    confidence: float
222    evidence: Dict[str, Any]
223    issued_at: str
224    expires_at: Optional[str]
225    nonce: str
226    signature: str = ""
227 
228    def unsigned_payload(self) -> Dict[str, Any]:
229        return {
230            "protocol": PROTOCOL_NAME,
231            "version": PROTOCOL_VERSION,
232            "id": self.id,
233            "issuer": self.issuer,
234            "subject": self.subject,
235            "kind": self.kind,
236            "score": self.score,
237            "confidence": self.confidence,
238            "evidence": self.evidence,
239            "issued_at": self.issued_at,
240            "expires_at": self.expires_at,
241            "nonce": self.nonce,
242        }
243 
244    def signing_bytes(self) -> bytes:
245        return canonical_json(self.unsigned_payload()).encode("utf-8")
246 
247    def sign(self, issuer_agent: Agent) -> None:
248        if issuer_agent.did != self.issuer:
249            raise CryptoError("issuer private key does not match claim issuer")
250        self.signature = CRYPTO.sign(issuer_agent.private_key, self.signing_bytes())
251 
252    def verify(self, issuer_agent: Agent) -> bool:
253        return CRYPTO.verify(
254            issuer_agent.public_key,
255            self.signing_bytes(),
256            self.signature,
257            private_key_for_hmac_fallback=issuer_agent.private_key,
258        )
259 
260    def to_dict(self) -> Dict[str, Any]:
261        payload = self.unsigned_payload()
262        payload["signature"] = self.signature
263        return payload
264 
265    @classmethod
266    def from_dict(cls, d: Dict[str, Any]) -> "TrustClaim":
267        return cls(
268            id=d["id"],
269            issuer=d["issuer"],
270            subject=d["subject"],
271            kind=d["kind"],
272            score=float(d["score"]),
273            confidence=float(d.get("confidence", 1.0)),
274            evidence=dict(d.get("evidence", {})),
275            issued_at=d["issued_at"],
276            expires_at=d.get("expires_at"),
277            nonce=d["nonce"],
278            signature=d.get("signature", ""),
279        )
280 
281 
282@dataclasses.dataclass
283class LedgerEvent:
284    index: int
285    timestamp: str
286    event_type: str
287    payload_hash: str
288    payload: Dict[str, Any]
289    previous_hash: str
290    event_hash: str
291 
292    @classmethod
293    def create(
294        cls,
295        index: int,
296        event_type: str,
297        payload: Dict[str, Any],
298        previous_hash: str,
299    ) -> "LedgerEvent":
300        base = {
301            "index": index,
302            "timestamp": now_iso(),
303            "event_type": event_type,
304            "payload_hash": digest(payload),
305            "payload": payload,
306            "previous_hash": previous_hash,
307        }
308        event_hash = digest(base)
309        return cls(event_hash=event_hash, **base)
310 
311    def to_dict(self) -> Dict[str, Any]:
312        return dataclasses.asdict(self)
313 
314    @classmethod
315    def from_dict(cls, d: Dict[str, Any]) -> "LedgerEvent":
316        return cls(**d)
317 
318 
319# ---------------------------------------------------------------------------
320# BAINK protocol engine
321# ---------------------------------------------------------------------------
322 
323class BainkProtocol:
324    def __init__(self) -> None:
325        self.agents: Dict[str, Agent] = {}
326        self.name_index: Dict[str, str] = {}
327        self.claims: Dict[str, TrustClaim] = {}
328        self.ledger: List[LedgerEvent] = []
329 
330    def add_event(self, event_type: str, payload: Dict[str, Any]) -> LedgerEvent:
331        previous_hash = self.ledger[-1].event_hash if self.ledger else GENESIS
332        event = LedgerEvent.create(len(self.ledger), event_type, payload, previous_hash)
333        self.ledger.append(event)
334        return event
335 
336    def create_agent(self, name: str, metadata: Optional[Dict[str, Any]] = None) -> Agent:
337        if name in self.name_index:
338            raise ValueError(f"agent name already exists: {name}")
339        agent = Agent.create(name, metadata)
340        self.agents[agent.did] = agent
341        self.name_index[name] = agent.did
342        self.add_event("AGENT_CREATED", agent.public_view())
343        return agent
344 
345    def resolve_agent(self, name_or_did: str) -> Agent:
346        did = self.name_index.get(name_or_did, name_or_did)
347        if did not in self.agents:
348            raise KeyError(f"unknown agent: {name_or_did}")
349        return self.agents[did]
350 
351    def issue_claim(
352        self,
353        issuer: str,
354        subject: str,
355        kind: str,
356        score: float,
357        confidence: float = 1.0,
358        evidence: Optional[Dict[str, Any]] = None,
359        expires_at: Optional[str] = None,
360    ) -> TrustClaim:
361        if not 0.0 <= score <= 1.0:
362            raise ValueError("score must be between 0.0 and 1.0")
363        if not 0.0 <= confidence <= 1.0:
364            raise ValueError("confidence must be between 0.0 and 1.0")
365 
366        issuer_agent = self.resolve_agent(issuer)
367        subject_agent = self.resolve_agent(subject)
368 
369        claim = TrustClaim(
370            id="claim_" + uuid.uuid4().hex,
371            issuer=issuer_agent.did,
372            subject=subject_agent.did,
373            kind=kind,
374            score=round(float(score), 6),
375            confidence=round(float(confidence), 6),
376            evidence=evidence or {},
377            issued_at=now_iso(),
378            expires_at=expires_at,
379            nonce=secrets.token_hex(16),
380        )
381        claim.sign(issuer_agent)
382        self.claims[claim.id] = claim
383        self.add_event("TRUST_CLAIM_ISSUED", claim.to_dict())
384        return claim
385 
386    def verify_claim(self, claim: TrustClaim) -> Tuple[bool, str]:
387        if claim.issuer not in self.agents:
388            return False, "unknown issuer"
389        if claim.subject not in self.agents:
390            return False, "unknown subject"
391        if claim.expires_at:
392            try:
393                exp = _dt.datetime.fromisoformat(claim.expires_at)
394                if exp.tzinfo is None:
395                    exp = exp.replace(tzinfo=_dt.timezone.utc)
396                if exp < _dt.datetime.now(_dt.timezone.utc):
397                    return False, "claim expired"
398            except Exception:
399                return False, "bad expires_at timestamp"
400        issuer_agent = self.agents[claim.issuer]
401        if not claim.verify(issuer_agent):
402            return False, "bad signature"
403        return True, "valid"
404 
405    def verify_ledger(self) -> Tuple[bool, List[str]]:
406        errors: List[str] = []
407        previous = GENESIS
408        for i, event in enumerate(self.ledger):
409            if event.index != i:
410                errors.append(f"ledger index mismatch at {i}")
411            if event.previous_hash != previous:
412                errors.append(f"previous hash mismatch at event {i}")
413            raw = event.to_dict().copy()
414            claimed_hash = raw.pop("event_hash")
415            recomputed = digest(raw)
416            if claimed_hash != recomputed:
417                errors.append(f"event hash mismatch at event {i}")
418            previous = event.event_hash
419 
420        for claim in self.claims.values():
421            ok, reason = self.verify_claim(claim)
422            if not ok:
423                errors.append(f"claim {claim.id}: {reason}")
424 
425        return not errors, errors
426 
427    def reputation(self, subject: str, kind: Optional[str] = None) -> Dict[str, Any]:
428        subject_agent = self.resolve_agent(subject)
429        valid_claims: List[TrustClaim] = []
430 
431        for claim in self.claims.values():
432            if claim.subject != subject_agent.did:
433                continue
434            if kind and claim.kind != kind:
435                continue
436            ok, _ = self.verify_claim(claim)
437            if ok:
438                valid_claims.append(claim)
439 
440        if not valid_claims:
441            return {
442                "subject": subject_agent.name,
443                "did": subject_agent.did,
444                "kind": kind or "*",
445                "score": None,
446                "claims": 0,
447                "explanation": "no valid claims",
448            }
449 
450        # Recursive-ish trust weighting:
451        # An issuer's influence is partly based on how much the network trusts
452        # the issuer. This is intentionally bounded and simple.
453        issuer_weights: Dict[str, float] = {}
454        for claim in valid_claims:
455            if claim.issuer == subject_agent.did:
456                issuer_weights[claim.issuer] = 0.05
457            else:
458                issuer_weights[claim.issuer] = 1.0
459 
460        weighted_sum = 0.0
461        total_weight = 0.0
462        for claim in valid_claims:
463            age_weight = self._age_decay(claim.issued_at)
464            issuer_weight = issuer_weights.get(claim.issuer, 1.0)
465            weight = claim.confidence * age_weight * issuer_weight
466            weighted_sum += claim.score * weight
467            total_weight += weight
468 
469        score = weighted_sum / total_weight if total_weight else 0.0
470        return {
471            "subject": subject_agent.name,
472            "did": subject_agent.did,
473            "kind": kind or "*",
474            "score": round(score, 6),
475            "claims": len(valid_claims),
476            "confidence_mass": round(total_weight, 6),
477            "interpretation": self._interpret(score),
478        }
479 
480    @staticmethod
481    def _age_decay(issued_at: str, half_life_days: float = 365.0) -> float:
482        try:
483            t = _dt.datetime.fromisoformat(issued_at)
484            if t.tzinfo is None:
485                t = t.replace(tzinfo=_dt.timezone.utc)
486            days = (_dt.datetime.now(_dt.timezone.utc) - t).total_seconds() / 86400.0
487            return 0.5 ** max(days, 0.0) / half_life_days
488        except Exception:
489            return 1.0
490 
491    @staticmethod
492    def _interpret(score: float) -> str:
493        if score >= 0.85:
494            return "high trust"
495        if score >= 0.65:
496            return "moderate trust"
497        if score >= 0.45:
498            return "uncertain trust"
499        return "low trust"
500 
501    def export(self) -> Dict[str, Any]:
502        return {
503            "protocol": PROTOCOL_NAME,
504            "version": PROTOCOL_VERSION,
505            "crypto_mode": CRYPTO.mode,
506            "agents": [dataclasses.asdict(a) for a in self.agents.values()],
507            "claims": [c.to_dict() for c in self.claims.values()],
508            "ledger": [e.to_dict() for e in self.ledger],
509        }
510 
511    @classmethod
512    def import_data(cls, data: Dict[str, Any]) -> "BainkProtocol":
513        p = cls()
514        for a in data.get("agents", []):
515            agent = Agent(**a)
516            p.agents[agent.did] = agent
517            p.name_index[agent.name] = agent.did
518        for c in data.get("claims", []):
519            claim = TrustClaim.from_dict(c)
520            p.claims[claim.id] = claim
521        for e in data.get("ledger", []):
522            p.ledger.append(LedgerEvent.from_dict(e))
523        return p
524 
525    def save(self, path: str) -> None:
526        Path(path).write_text(json.dumps(self.export(), indent=2), encoding="utf-8")
527 
528    @classmethod
529    def load(cls, path: str) -> "BainkProtocol":
530        if not Path(path).exists():
531            return cls()
532        return cls.import_data(json.loads(Path(path).read_text(encoding="utf-8")))
533 
534 
535# ---------------------------------------------------------------------------
536# CLI
537# ---------------------------------------------------------------------------
538 
539def print_json(obj: Any) -> None:
540    print(json.dumps(obj, indent=2, ensure_ascii=False))
541 
542 
543def cmd_demo(_: argparse.Namespace) -> None:
544    p = BainkProtocol()
545    alice = p.create_agent("Alice", {"role": "auditor"})
546    bob = p.create_agent("Bob", {"role": "builder"})
547    carol = p.create_agent("Carol", {"role": "customer"})
548    p.issue_claim("Alice", "Bob", "reliability", 0.92, 0.95, {"source": "delivery audit"})
549    p.issue_claim("Carol", "Bob", "integrity", 0.88, 0.80, {"source": "completed contract"})
550    p.issue_claim("Alice", "Carol", "payment_behavior", 0.97, 0.90, {"source": "invoice record"})
551 
552    ok, errors = p.verify_ledger()
553    print("BAINK Protocol demo")
554    print(f"crypto_mode: {CRYPTO.mode}")
555    print(f"ledger_valid: {ok}")
556    if errors:
557        print_json(errors)
558    print("\nAgents:")
559    print_json([a.public_view() for a in p.agents.values()])
560    print("\nBob reputation:")
561    print_json(p.reputation("Bob"))
562    print("\nLedger head:")
563    print_json(p.ledger[-1].to_dict())
564 
565 
566def cmd_init(args: argparse.Namespace) -> None:
567    p = BainkProtocol()
568    p.save(args.db)
569    print(f"initialized {args.db}")
570 
571 
572def cmd_create_agent(args: argparse.Namespace) -> None:
573    p = BainkProtocol.load(args.db)
574    metadata = json.loads(args.metadata) if args.metadata else {}
575    agent = p.create_agent(args.name, metadata)
576    p.save(args.db)
577    print_json(agent.public_view())
578 
579 
580def cmd_claim(args: argparse.Namespace) -> None:
581    p = BainkProtocol.load(args.db)
582    evidence = json.loads(args.evidence) if args.evidence else {}
583    claim = p.issue_claim(
584        issuer=args.issuer,
585        subject=args.subject,
586        kind=args.kind,
587        score=args.score,
588        confidence=args.confidence,
589        evidence=evidence,
590        expires_at=args.expires_at,
591    )
592    p.save(args.db)
593    print_json(claim.to_dict())
594 
595 
596def cmd_verify(args: argparse.Namespace) -> None:
597    p = BainkProtocol.load(args.db)
598    ok, errors = p.verify_ledger()
599    print_json({"valid": ok, "errors": errors})
600 
601 
602def cmd_reputation(args: argparse.Namespace) -> None:
603    p = BainkProtocol.load(args.db)
604    print_json(p.reputation(args.subject, args.kind))
605 
606 
607def cmd_agents(args: argparse.Namespace) -> None:
608    p = BainkProtocol.load(args.db)
609    print_json([a.public_view() for a in p.agents.values()])
610 
611 
612def cmd_ledger(args: argparse.Namespace) -> None:
613    p = BainkProtocol.load(args.db)
614    print_json([e.to_dict() for e in p.ledger])
615 
616 
617def build_parser() -> argparse.ArgumentParser:
618    parser = argparse.ArgumentParser(
619        prog="baink_protocol",
620        formatter_class=argparse.RawDescriptionHelpFormatter,
621        description=textwrap.dedent("""
622        BAINK Protocol — Engineering Trust Infrastructure
623 
624        A complete flattened executable trust layer in one Python file.
625        """),
626    )
627    sub = parser.add_subparsers(required=True)
628 
629    demo = sub.add_parser("demo")
630    demo.set_defaults(func=cmd_demo)
631 
632    init = sub.add_parser("init")
633    init.add_argument("--db", default="baink.json")
634    init.set_defaults(func=cmd_init)
635 
636    ca = sub.add_parser("create-agent")
637    ca.add_argument("name")
638    ca.add_argument("--metadata", help='JSON metadata, e.g. {"role":"auditor"}')
639    ca.add_argument("--db", default="baink.json")
640    ca.set_defaults(func=cmd_create_agent)
641 
642    claim = sub.add_parser("claim")
643    claim.add_argument("--issuer", required=True)
644    claim.add_argument("--subject", required=True)
645    claim.add_argument("--kind", required=True)
646    claim.add_argument("--score", type=float, required=True)
647    claim.add_argument("--confidence", type=float, default=1.0)
648    claim.add_argument("--evidence", help='JSON evidence object')
649    claim.add_argument("--expires-at")
650    claim.add_argument("--db", default="baink.json")
651    claim.set_defaults(func=cmd_claim)
652 
653    verify = sub.add_parser("verify")
654    verify.add_argument("--db", default="baink.json")
655    verify.set_defaults(func=cmd_verify)
656 
657    rep = sub.add_parser("reputation")
658    rep.add_argument("subject")
659    rep.add_argument("--kind")
660    rep.add_argument("--db", default="baink.json")
661    rep.set_defaults(func=cmd_reputation)
662 
663    agents = sub.add_parser("agents")
664    agents.add_argument("--db", default="baink.json")
665    agents.set_defaults(func=cmd_agents)
666 
667    ledger = sub.add_parser("ledger")
668    ledger.add_argument("--db", default="baink.json")
669    ledger.set_defaults(func=cmd_ledger)
670 
671    return parser
672 
673 
674def main(argv: Optional[List[str]] = None) -> int:
675    parser = build_parser()
676    args = parser.parse_args(argv)
677    try:
678        args.func(args)
679        return 0
680    except Exception as exc:
681        print(f"BAINK_ERROR: {exc}", file=sys.stderr)
682        return 1
683 
684 
685if __name__ == "__main__":
686    raise SystemExit(main())