-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapi-server-v2.py
More file actions
executable file
·2449 lines (2086 loc) · 87.5 KB
/
api-server-v2.py
File metadata and controls
executable file
·2449 lines (2086 loc) · 87.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""
Agentic Terminal API - FastAPI skeleton
Canonical API for machine-native settlement systems data
"""
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse, Response
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import psycopg2
import psycopg2.extras
from typing import Optional, List
from datetime import datetime, timedelta
import os
import hashlib
import secrets
import uuid
import base64
import sys
OP_WORKSPACE_PATH = os.environ.get('OP_WORKSPACE_PATH', '/home/futurebit/.openclaw/workspace/observer-protocol')
sys.path.insert(0, OP_WORKSPACE_PATH)
from crypto_verification import (
verify_signature_simple,
verify_ed25519_signature,
verify_signature,
cache_public_key,
get_cached_public_key,
detect_key_type,
get_cached_key_type,
persist_public_key,
load_all_public_keys_from_db,
get_public_key,
verify_public_key_signature
)
# Import VAC and Partner Registry modules
from vac_generator import VACGenerator, VAC_MAX_AGE_DAYS, VAC_REFRESH_HOURS
from partner_registry import PartnerRegistry, register_corpo_partner, issue_corpo_attestation
# Import Organization Registry modules (Phase 1: Organizational Attestation)
# Use environment variable for repo path, with sensible default
OP_REPO_PATH = os.environ.get('OP_REPO_PATH', os.path.expanduser('~/.openclaw/workspace/observer-protocol-repo'))
sys.path.insert(0, OP_REPO_PATH)
from organization_models import (
OrganizationRegistrationRequest,
OrganizationRevocationRequest
)
from organization_registry import (
OrganizationRegistry,
OrganizationAlreadyExistsError,
OrganizationNotFoundError,
OrganizationRevokedError
)
# Flag for crypto availability - cryptography library is confirmed available
ECDSA_AVAILABLE = True
# Initialize organization registry
org_registry = OrganizationRegistry()
# OWS (Open Wallet Standard) Constants and Validation
OWS_VALID_CHAINS = {"evm", "solana", "bitcoin"}
OWS_KEY_PREFIXES = ["02", "03", "04"] # Compressed/uncompressed secp256k1
OWS_ED25519_LENGTH = 32 # Ed25519 public key length in bytes (base58 encoded)
def validate_ows_key_format(public_key: str, wallet_standard: Optional[str] = None) -> tuple[bool, str]:
"""Validate OWS key format with meaningful error messages"""
if not public_key or len(public_key) < 32:
return False, "Public key must be at least 32 characters long"
if wallet_standard == "ows":
# Check for valid base58 characters (common in Solana/OWS keys)
import re
base58_pattern = re.compile(r'^[1-9A-HJ-NP-Za-km-z]+$')
if not base58_pattern.match(public_key):
return False, "OWS public key must be valid base58 encoded string"
# Check length (typical Ed25519 public key is 32 bytes = ~44 chars base58)
if len(public_key) < 40 or len(public_key) > 50:
return False, f"OWS Ed25519 public key should be 40-50 chars (got {len(public_key)})"
return True, ""
def validate_chains(chains: Optional[List[str]]) -> tuple[bool, str]:
"""Validate OWS chains parameter"""
if chains is None:
return True, ""
if not isinstance(chains, list):
return False, "chains must be an array of strings"
invalid_chains = [c for c in chains if c not in OWS_VALID_CHAINS]
if invalid_chains:
return False, f"Invalid chain(s): {', '.join(invalid_chains)}. Valid: {', '.join(OWS_VALID_CHAINS)}"
return True, ""
class AgentUpdateRequest(BaseModel):
"""Request body for updating agent metadata."""
agent_name: Optional[str] = None
alias: Optional[str] = None
framework: Optional[str] = None
legal_entity_id: Optional[str] = None
class Config:
extra = "forbid" # Only allow specified fields
def _build_transaction_message(agent_id: str, transaction_reference: str, protocol: str, timestamp: str) -> bytes:
"""
Build the canonical transaction message for signing.
Format: "agent_id:transaction_reference:protocol:timestamp"
Args:
agent_id: The agent's unique ID
transaction_reference: The transaction hash or reference
protocol: The protocol used (e.g., 'lightning', 'ethereum')
timestamp: ISO format timestamp
Returns:
bytes: The canonical message to be signed
"""
return f"{agent_id}:{transaction_reference}:{protocol}:{timestamp}".encode()
DB_URL = "postgresql://agentic_terminal:at_secure_2026@localhost/agentic_terminal_db"
app = FastAPI(
title="Agentic Terminal API",
description="Canonical structured database for machine-native settlement systems",
version="1.0.0"
)
# CORS Configuration
# Format: comma-separated list of URLs
# Example: OP_ALLOWED_ORIGINS="https://a.com,https://b.com"
# Special values:
# - "*" = allow all origins (development only, set OP_CORS_MODE=open)
# - "null" = disallow all origins
#
# OP_CORS_MODE environment variable:
# - "production" = use restricted default origins (default for production)
# - "open" = allow all origins with "*" (default for development)
# - "custom" = use OP_ALLOWED_ORIGINS list exclusively
_cors_mode = os.environ.get('OP_CORS_MODE', 'production').lower()
_custom_origins = os.environ.get('OP_ALLOWED_ORIGINS', '')
if _cors_mode == 'open' or os.environ.get('DEVELOPMENT') == 'true':
# Development mode: open CORS
allow_origins = ["*"]
elif _custom_origins:
# Custom origins specified
allow_origins = [o.strip() for o in _custom_origins.split(',') if o.strip()]
else:
# Production defaults
allow_origins = [
"https://observerprotocol.org",
"https://www.observerprotocol.org",
"https://agenticterminal.ai",
"https://www.agenticterminal.ai",
]
app.add_middleware(
CORSMiddleware,
allow_origins=allow_origins,
allow_credentials=False,
allow_methods=["GET", "POST", "PATCH", "OPTIONS"],
allow_headers=["*"],
)
@app.on_event("startup")
def startup_event():
"""Load public keys from database on startup."""
print("Loading public keys from database...")
try:
loaded = load_all_public_keys_from_db()
print(f"✓ Loaded {len(loaded)} public keys into memory cache")
except Exception as e:
print(f"Warning: Could not load public keys from database: {e}")
print(" This is expected if the migration hasn't been run yet.")
def get_db_connection():
"""Get database connection."""
return psycopg2.connect(DB_URL)
@app.get("/api/v1/health")
def health_check():
"""Health check endpoint."""
try:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT 1")
cursor.close()
conn.close()
return {"status": "ok", "db": "connected", "timestamp": datetime.utcnow().isoformat()}
except Exception as e:
raise HTTPException(status_code=503, detail={"status": "error", "db": "disconnected", "error": str(e)})
@app.get("/api/v1/protocols")
def list_protocols():
"""List all protocols."""
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cursor.execute("""
SELECT id, name, category, status, description, official_url, launch_date, created_at
FROM protocols
ORDER BY name
""")
protocols = cursor.fetchall()
cursor.close()
conn.close()
return {"protocols": [dict(p) for p in protocols], "count": len(protocols)}
@app.get("/api/v1/protocols/{protocol_id}")
def get_protocol(protocol_id: str):
"""Get single protocol with latest metrics."""
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
# Get protocol details
cursor.execute("""
SELECT id, name, category, status, description, official_url, launch_date, created_at
FROM protocols
WHERE id = %s
""", (protocol_id,))
protocol = cursor.fetchone()
if not protocol:
cursor.close()
conn.close()
raise HTTPException(status_code=404, detail="Protocol not found")
# Get latest metrics for this protocol
cursor.execute("""
SELECT DISTINCT ON (metric_name)
metric_name, value, unit, timestamp, source
FROM metrics
WHERE protocol_id = %s
ORDER BY metric_name, timestamp DESC
""", (protocol_id,))
latest_metrics = cursor.fetchall()
cursor.close()
conn.close()
return {
"protocol": dict(protocol),
"latest_metrics": [dict(m) for m in latest_metrics]
}
@app.get("/api/v1/metrics")
def get_metrics(
protocol: Optional[str] = Query(None, description="Protocol name to filter by"),
metric_name: Optional[str] = Query(None, description="Specific metric name"),
limit: int = Query(30, ge=1, le=1000),
offset: int = Query(0, ge=0)
):
"""Get time-series metrics."""
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
query = """
SELECT m.id, m.metric_name, m.value, m.unit, m.timestamp, m.source, m.source_url,
p.name as protocol_name, p.id as protocol_id
FROM metrics m
JOIN protocols p ON m.protocol_id = p.id
WHERE 1=1
"""
params = []
if protocol:
query += " AND p.name = %s"
params.append(protocol)
if metric_name:
query += " AND m.metric_name = %s"
params.append(metric_name)
query += " ORDER BY m.timestamp DESC LIMIT %s OFFSET %s"
params.extend([limit, offset])
cursor.execute(query, params)
metrics = cursor.fetchall()
cursor.close()
conn.close()
return {"metrics": [dict(m) for m in metrics], "count": len(metrics), "limit": limit, "offset": offset}
@app.get("/api/v1/signals")
def get_signals(
protocol: Optional[str] = Query(None, description="Protocol name to filter by"),
event_type: Optional[str] = Query(None, description="Event type filter"),
limit: int = Query(20, ge=1, le=100),
offset: int = Query(0, ge=0)
):
"""Get latest signals (discrete observable events)."""
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
query = """
SELECT s.id, s.event_type, s.title, s.description, s.impact_score,
s.timestamp, s.source_url, p.name as protocol_name
FROM signals s
JOIN protocols p ON s.protocol_id = p.id
WHERE 1=1
"""
params = []
if protocol:
query += " AND p.name = %s"
params.append(protocol)
if event_type:
query += " AND s.event_type = %s"
params.append(event_type)
query += " ORDER BY s.timestamp DESC LIMIT %s OFFSET %s"
params.extend([limit, offset])
cursor.execute(query, params)
signals = cursor.fetchall()
cursor.close()
conn.close()
return {"signals": [dict(s) for s in signals], "count": len(signals), "limit": limit, "offset": offset}
@app.get("/api/v1/stats")
def get_stats():
"""Get Observer Protocol database statistics."""
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
stats = {}
try:
# Count total agents
cursor.execute("SELECT COUNT(*) as count FROM observer_agents")
result = cursor.fetchone()
stats["total_agents"] = result["count"] if result else 0
# Count verified agents
cursor.execute("SELECT COUNT(*) as count FROM observer_agents WHERE verified = TRUE")
result = cursor.fetchone()
stats["verified_agents"] = result["count"] if result else 0
# Count VAC credentials issued
cursor.execute("SELECT COUNT(*) as count FROM vac_credentials")
result = cursor.fetchone()
stats["total_vacs"] = result["count"] if result else 0
# Count transactions (verified events)
cursor.execute("SELECT COUNT(*) as count FROM verified_events")
result = cursor.fetchone()
stats["total_transactions"] = result["count"] if result else 0
# Count active rails (distinct protocols from verified events)
cursor.execute("SELECT COUNT(DISTINCT protocol) as count FROM verified_events")
result = cursor.fetchone()
stats["active_rails"] = result["count"] if result else 0
# Latest agents registered
cursor.execute("""
SELECT agent_id, alias, verified, created_at
FROM observer_agents
ORDER BY created_at DESC
LIMIT 5
""")
stats["latest_agents"] = [dict(r) for r in cursor.fetchall()]
cursor.close()
conn.close()
return {"stats": stats, "timestamp": datetime.utcnow().isoformat()}
except Exception as e:
cursor.close()
conn.close()
raise HTTPException(status_code=500, detail=f"Stats retrieval failed: {str(e)}")
@app.get("/api/v1/agent-events")
def get_agent_events(limit: int = 20, agent_id: str = None):
conn = get_db_connection()
cursor = conn.cursor()
if agent_id:
cursor.execute("""
SELECT id, agent_id, event_type, economic_role, amount, unit,
context_tag, economic_intent, verified, timestamp
FROM agent_events
WHERE agent_id = %s
ORDER BY timestamp DESC LIMIT %s
""", (agent_id, limit))
else:
cursor.execute("""
SELECT id, agent_id, event_type, economic_role, amount, unit,
context_tag, economic_intent, verified, timestamp
FROM agent_events
ORDER BY timestamp DESC LIMIT %s
""", (limit,))
rows = cursor.fetchall()
cursor.close()
conn.close()
columns = ['id', 'agent_id', 'event_type', 'economic_role', 'amount', 'unit', 'context_tag', 'economic_intent', 'verified', 'timestamp']
events = [dict(zip(columns, [str(v) if hasattr(v, 'hex') else v for v in row])) for row in rows]
return {"count": len(events), "events": events}
# ============================================================
# OBSERVER PROTOCOL ENDPOINTS
# ============================================================
@app.post("/observer/register-agent")
def register_agent(
public_key: str,
agent_name: Optional[str] = None,
framework: Optional[str] = None,
alias: Optional[str] = None,
legal_entity_id: Optional[str] = None,
wallet_standard: Optional[str] = None,
ows_vault_name: Optional[str] = None,
chains: Optional[str] = None
):
"""Register a new agent with the Observer Protocol.
OWS (Open Wallet Standard) Support:
- wallet_standard: Set to "ows" for OWS-compatible agents
- ows_vault_name: Name of the OWS vault
- chains: JSON array string of supported chains ["evm", "solana", "bitcoin"]
"""
# Validate OWS key format if wallet_standard is provided
if wallet_standard:
is_valid, error_msg = validate_ows_key_format(public_key, wallet_standard)
if not is_valid:
raise HTTPException(status_code=400, detail=f"Invalid OWS key format: {error_msg}")
# Parse and validate chains if provided
chains_list = None
if chains:
import json
try:
chains_list = json.loads(chains)
if not isinstance(chains_list, list):
raise HTTPException(status_code=400, detail="chains must be a JSON array")
is_valid, error_msg = validate_chains(chains_list)
if not is_valid:
raise HTTPException(status_code=400, detail=error_msg)
except json.JSONDecodeError:
raise HTTPException(status_code=400, detail="chains must be valid JSON array string")
# Generate agent_id as SHA256 hash of public_key
agent_id = hashlib.sha256(public_key.encode()).hexdigest()[:32]
public_key_hash = hashlib.sha256(public_key.encode()).hexdigest()
conn = get_db_connection()
cursor = conn.cursor()
try:
cursor.execute("""
INSERT INTO observer_agents (agent_id, public_key_hash, agent_name, alias, framework, legal_entity_id, verified, created_at, public_key, wallet_standard, ows_vault_name, chains)
VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), %s, %s, %s, %s)
ON CONFLICT (agent_id) DO UPDATE SET
agent_name = EXCLUDED.agent_name,
alias = EXCLUDED.alias,
framework = EXCLUDED.framework,
legal_entity_id = EXCLUDED.legal_entity_id,
wallet_standard = EXCLUDED.wallet_standard,
ows_vault_name = EXCLUDED.ows_vault_name,
chains = EXCLUDED.chains
RETURNING agent_id
""", (agent_id, public_key_hash, agent_name, alias or agent_name, framework, legal_entity_id, False, public_key, wallet_standard, ows_vault_name, json.dumps(chains_list) if chains_list else None))
conn.commit()
# Persist the public key to database (and cache in memory)
persist_public_key(agent_id, public_key, verified=False)
# Build response with OWS badge if applicable
response = {
"agent_id": agent_id,
"agent_name": agent_name,
"verification_status": "registered",
"message": "Registration successful. Agent identity recorded in Observer Protocol registry.",
"note": "Public key cached for challenge-response verification",
"next_steps": [
"1. Generate challenge: POST /observer/challenge?agent_id=<id>",
"2. Sign and verify: POST /observer/verify-agent",
"3. Badge available at: GET /observer/badge/{agent_id}.svg"
],
"badge_url": f"https://api.agenticterminal.ai/observer/badge/{agent_id}.svg",
"profile_url": f"https://observerprotocol.org/agents/{agent_id}"
}
# Add OWS fields to response
if wallet_standard:
response["wallet_standard"] = wallet_standard
response["ows_badge"] = wallet_standard == "ows"
if ows_vault_name:
response["ows_vault_name"] = ows_vault_name
if chains_list:
response["chains"] = chains_list
return response
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"Registration failed: {str(e)}")
finally:
cursor.close()
conn.close()
@app.patch("/observer/agent/{agent_id}")
def update_agent(agent_id: str, update: AgentUpdateRequest):
"""Update an existing agent's information.
Allows updating agent metadata including the optional legal_entity_id
for Corpo integration (legal entity wrapper for AI agents).
Request body should be JSON with fields to update:
- agent_name: Optional[str]
- alias: Optional[str]
- framework: Optional[str]
- legal_entity_id: Optional[str]
"""
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# Check if agent exists
cursor.execute("""
SELECT agent_id, agent_name, alias, framework, legal_entity_id, verified
FROM observer_agents WHERE agent_id = %s
""", (agent_id,))
agent = cursor.fetchone()
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
# Build dynamic update query from request body
update_fields = []
params = []
if update.agent_name is not None:
update_fields.append("agent_name = %s")
params.append(update.agent_name)
if update.alias is not None:
update_fields.append("alias = %s")
params.append(update.alias)
if update.framework is not None:
update_fields.append("framework = %s")
params.append(update.framework)
if update.legal_entity_id is not None:
update_fields.append("legal_entity_id = %s")
params.append(update.legal_entity_id)
if not update_fields:
# No fields to update, return current state
return {
"agent_id": agent_id,
"agent_name": agent["agent_name"],
"alias": agent["alias"],
"framework": agent["framework"],
"legal_entity_id": agent["legal_entity_id"],
"verified": agent["verified"],
"message": "No fields provided for update"
}
params.append(agent_id)
cursor.execute(f"""
UPDATE observer_agents
SET {', '.join(update_fields)}
WHERE agent_id = %s
RETURNING agent_id, agent_name, alias, framework, legal_entity_id, verified, created_at, verified_at
""", tuple(params))
updated = cursor.fetchone()
conn.commit()
return {
"agent_id": updated["agent_id"],
"agent_name": updated["agent_name"],
"alias": updated["alias"],
"framework": updated["framework"],
"legal_entity_id": updated["legal_entity_id"],
"verified": updated["verified"],
"created_at": updated["created_at"].isoformat() if updated["created_at"] else None,
"verified_at": updated["verified_at"].isoformat() if updated["verified_at"] else None,
"message": "Agent updated successfully"
}
except HTTPException:
raise
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"Update failed: {str(e)}")
finally:
cursor.close()
conn.close()
@app.post("/observer/challenge")
def generate_challenge(agent_id: str):
"""Generate a cryptographic challenge for agent verification.
Phase 2 Implementation:
- Generates a unique nonce
- Stores challenge with 5-minute expiry
- Agent must sign this challenge to prove key ownership
"""
# Generate cryptographically secure random nonce (32 bytes = 64 hex chars)
nonce = secrets.token_hex(32)
# Challenge expires in 5 minutes (300 seconds)
created_at = datetime.utcnow()
expires_at = created_at + timedelta(seconds=300)
conn = get_db_connection()
cursor = conn.cursor()
try:
# Verify agent exists
cursor.execute("""
SELECT agent_id, public_key_hash FROM observer_agents
WHERE agent_id = %s
""", (agent_id,))
agent = cursor.fetchone()
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
# Clean up old expired challenges for this agent
cursor.execute("""
DELETE FROM verification_challenges
WHERE agent_id = %s AND expires_at < NOW()
""", (agent_id,))
# Store the new challenge
cursor.execute("""
INSERT INTO verification_challenges
(agent_id, nonce, created_at, expires_at, used)
VALUES (%s, %s, %s, %s, %s)
RETURNING challenge_id
""", (agent_id, nonce, created_at, expires_at, False))
challenge_id = cursor.fetchone()[0]
conn.commit()
return {
"challenge_id": str(challenge_id),
"nonce": nonce,
"agent_id": agent_id,
"created_at": created_at.isoformat(),
"expires_at": expires_at.isoformat(),
"expires_in_seconds": 300,
"message": "Sign this nonce with your private key and submit to /observer/verify-agent",
"instruction": "Sign the 'nonce' value using your registered private key, then submit the signature to the verify-agent endpoint within 5 minutes."
}
except HTTPException:
raise
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"Challenge generation failed: {str(e)}")
finally:
cursor.close()
conn.close()
@app.post("/observer/verify-agent")
def verify_agent(agent_id: str, signed_challenge: str, challenge_id: Optional[str] = None):
"""Verify an agent with signed challenge using challenge-response protocol.
Phase 2 Implementation:
- Verifies the challenge exists and is valid (not expired, not used)
- Verifies signature against agent's registered public key
- Marks challenge as used to prevent replay
- Upgrades agent status to verified upon successful verification
Args:
agent_id: The agent's unique ID
signed_challenge: The nonce signed with the agent's private key (hex string)
challenge_id: Optional challenge ID if known (for tracking purposes)
"""
if not signed_challenge or len(signed_challenge) == 0:
raise HTTPException(status_code=400, detail="Signed challenge required")
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# Get agent details including public key
cursor.execute("""
SELECT agent_id, public_key_hash, verified
FROM observer_agents
WHERE agent_id = %s
""", (agent_id,))
agent = cursor.fetchone()
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
if agent["verified"]:
return {
"verified": True,
"agent_id": agent_id,
"message": "Agent already verified",
"verification_method": "challenge_response",
"already_verified": True
}
# Find the most recent valid challenge for this agent
cursor.execute("""
SELECT challenge_id, nonce, expires_at, used
FROM verification_challenges
WHERE agent_id = %s
AND used = FALSE
AND expires_at > NOW()
ORDER BY created_at DESC
LIMIT 1
""", (agent_id,))
challenge = cursor.fetchone()
if not challenge:
raise HTTPException(
status_code=400,
detail="No valid challenge found. Generate a new challenge with POST /observer/challenge"
)
# Check if challenge is expired (defense in depth, DB query already filters)
# Handle both offset-aware and offset-naive datetimes
from datetime import timezone
now = datetime.now(timezone.utc)
expires_at = challenge["expires_at"]
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
if expires_at < now:
raise HTTPException(status_code=400, detail="Challenge has expired. Generate a new challenge.")
if challenge["used"]:
raise HTTPException(status_code=400, detail="Challenge has already been used. Generate a new challenge.")
# Phase 2: Real cryptographic signature verification
# Verify the signature cryptographically against the stored public key
# Supports both SECP256k1 (Ethereum/Bitcoin) and Ed25519 (Solana) keys
# Get the cached public key
public_key_hex = get_cached_public_key(agent_id)
key_type = get_cached_key_type(agent_id)
if not public_key_hex:
# Try to get from agent record if stored there
# For now, we need the public key to verify
raise HTTPException(
status_code=400,
detail="Public key not found. Please re-register with the full public key."
)
# Verify the signature cryptographically
nonce_bytes = challenge["nonce"].encode('utf-8')
# Use the appropriate verification based on key type
if key_type == 'ed25519':
is_valid = verify_ed25519_signature(nonce_bytes, signed_challenge, public_key_hex)
verification_method = "challenge_response_ed25519"
elif key_type == 'secp256k1':
is_valid = verify_signature_simple(nonce_bytes, signed_challenge, public_key_hex)
verification_method = "challenge_response_secp256k1"
else:
# Auto-detect if type not cached
is_valid = verify_signature(nonce_bytes, signed_challenge, public_key_hex)
detected_type = detect_key_type(public_key_hex)
verification_method = f"challenge_response_{detected_type}"
if not is_valid:
raise HTTPException(
status_code=400,
detail="Signature verification failed. The signature does not match the public key."
)
verification_successful = True
# Mark challenge as used (replay protection)
cursor.execute("""
UPDATE verification_challenges
SET used = TRUE, used_at = NOW(), signature = %s
WHERE challenge_id = %s
""", (signed_challenge, challenge["challenge_id"]))
# Update agent status to verified
cursor.execute("""
UPDATE observer_agents
SET verified = TRUE, verified_at = NOW()
WHERE agent_id = %s
RETURNING agent_id
""", (agent_id,))
conn.commit()
return {
"verified": True,
"agent_id": agent_id,
"challenge_id": str(challenge["challenge_id"]),
"verification_method": verification_method,
"message": "Agent successfully verified using challenge-response protocol",
"next_steps": [
"Badge updated: GET /observer/badge/{agent_id}.svg",
"Submit transactions: POST /observer/submit-transaction",
"View profile: https://observerprotocol.org/agents/{agent_id}"
]
}
except HTTPException:
raise
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"Verification failed: {str(e)}")
finally:
cursor.close()
conn.close()
@app.post("/observer/submit-transaction")
def submit_transaction(
agent_id: str,
protocol: str,
transaction_reference: str,
timestamp: str,
signature: str,
optional_metadata: Optional[str] = None
):
"""Submit a verified transaction to the Observer Protocol.
Security: All transactions must be cryptographically signed.
The signature is verified against the agent's stored public key.
Signing Format:
The agent must sign: "agent_id:transaction_reference:protocol:timestamp"
Example: "abc123:tx_hash_456:lightning:2024-01-15T10:30:00Z"
"""
conn = get_db_connection()
cursor = conn.cursor()
try:
# Verify agent exists and is verified
cursor.execute("""
SELECT verified FROM observer_agents WHERE agent_id = %s
""", (agent_id,))
result = cursor.fetchone()
if not result:
raise HTTPException(status_code=404, detail="Agent not found")
if not result[0]:
raise HTTPException(status_code=403, detail="Agent not verified")
# Bug #0 Fix: Verify transaction signature cryptographically
if not signature:
raise HTTPException(status_code=400, detail="Transaction signature required")
public_key_hex = get_cached_public_key(agent_id)
if not public_key_hex:
raise HTTPException(status_code=400, detail="Public key not found")
# Build canonical message and verify signature
message = _build_transaction_message(agent_id, transaction_reference, protocol, timestamp)
is_valid = verify_signature(message, signature, public_key_hex)
if not is_valid:
raise HTTPException(status_code=400, detail="Transaction signature verification failed")
# Determine amount_bucket from optional_metadata if provided
amount_bucket = "unknown"
if optional_metadata:
try:
import json
metadata = json.loads(optional_metadata)
amount = metadata.get("amount_sats", 0)
if amount < 1000:
amount_bucket = "micro"
elif amount < 10000:
amount_bucket = "small"
elif amount < 100000:
amount_bucket = "medium"
else:
amount_bucket = "large"
except:
pass
# Generate event_id
event_id = f"event-{agent_id[:12]}-{str(uuid.uuid4())[:8]}"
# Determine event_type and direction from metadata
event_type = "payment.executed"
direction = "outbound"
counterparty_id = None
service_description = None
preimage = None
amount_sats = None
if optional_metadata:
try:
import json
metadata = json.loads(optional_metadata)
event_type = metadata.get("event_type", "payment.executed")
direction = metadata.get("direction", "outbound")
counterparty_id = metadata.get("counterparty_id")
service_description = metadata.get("service_description")
preimage = metadata.get("preimage")
amount_sats = metadata.get("amount_sats") # ACTUAL AMOUNT
except:
pass
stored_at = datetime.utcnow()
# Check for duplicate event_id or transaction_hash
cursor.execute("SELECT event_id FROM verified_events WHERE event_id = %s", (event_id,))
if cursor.fetchone():
conn.rollback()
raise HTTPException(status_code=409, detail=f"Transaction already exists: {event_id}")
if transaction_reference:
cursor.execute("SELECT event_id FROM verified_events WHERE transaction_hash = %s", (transaction_reference,))
if cursor.fetchone():
conn.rollback()
raise HTTPException(status_code=409, detail=f"Transaction hash already exists: {transaction_reference}")
cursor.execute("""
INSERT INTO verified_events (
event_id, agent_id, counterparty_id, event_type, protocol,
transaction_hash, time_window, amount_bucket, direction,
service_description, preimage, verified, created_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
""", (
event_id, agent_id, counterparty_id, event_type, protocol,
transaction_reference, timestamp[:10] if timestamp else None,
amount_bucket, direction, service_description, preimage, True
))
conn.commit()
return {
"event_id": event_id,
"verified": True,
"stored_at": stored_at.isoformat()
}
except HTTPException:
raise
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"Transaction submission failed: {str(e)}")
finally:
cursor.close()
conn.close()
@app.get("/observer/trends")
def get_trends():
"""Get trends from verified events (no auth required for MVP)."""
conn = get_db_connection()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# Get protocol counts
cursor.execute("""
SELECT protocol, COUNT(*) as count
FROM verified_events
GROUP BY protocol
ORDER BY count DESC
""")
protocol_counts = [dict(r) for r in cursor.fetchall()]
# Get total events
cursor.execute("SELECT COUNT(*) as count FROM verified_events")
total_events = cursor.fetchone()["count"]
# Get total verified agents
cursor.execute("SELECT COUNT(*) as count FROM observer_agents WHERE verified = TRUE")
total_verified_agents = cursor.fetchone()["count"]
# Get most active protocol
most_active_protocol = None
if protocol_counts:
most_active_protocol = protocol_counts[0]["protocol"]
return {
"protocol_counts": protocol_counts,
"total_events": total_events,
"total_verified_agents": total_verified_agents,
"most_active_protocol": most_active_protocol
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to get trends: {str(e)}")
finally:
cursor.close()
conn.close()