Skip to content

Commit 50af836

Browse files
chgaoweichgaowei
committed
complete code
1 parent b63fab0 commit 50af836

File tree

5 files changed

+589
-3
lines changed

5 files changed

+589
-3
lines changed

‎.gitignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,3 +11,5 @@ examples/negotiation_mode/workflow_code/
1111
examples/negotiation_mode/protocol_code/
1212
examples/negotiation_mode/*.json
1313
examples/simple_node_mode/*.json
14+
examples/did_wba_examples/did_keys/user_*/
15+

‎agent_connect/authentication/did_wba.py‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -172,18 +172,23 @@ async def resolve_did_wba_document(did: str) -> Dict:
172172
async with aiohttp.ClientSession(timeout=timeout) as session:
173173
url = f"https://{domain}"
174174
if path_segments:
175-
url += '/' + '/'.join(path_segments)
175+
url += '/' + '/'.join(path_segments) + '/did.json'
176176
else:
177177
url += '/.well-known/did.json'
178-
178+
179179
logging.debug(f"Requesting DID document from URL: {url}")
180-
180+
181+
# TODO: Add DNS-over-HTTPS support
182+
# resolver = aiohttp.AsyncResolver(nameservers=['8.8.8.8'])
183+
# connector = aiohttp.TCPConnector(resolver=resolver)
184+
181185
async with session.get(
182186
url,
183187
headers={
184188
'Accept': 'application/json'
185189
},
186190
ssl=True
191+
# connector=connector
187192
) as response:
188193
response.raise_for_status()
189194
did_document = await response.json()

‎examples/did_wba_examples/basic.py‎

Lines changed: 201 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,201 @@
1+
# AgentConnect: https://github.com/chgaowei/AgentConnect
2+
# Author: GaoWei Chang
3+
# Email: chgaowei@gmail.com
4+
# Website: https://agent-network-protocol.com/
5+
#
6+
# This project is open-sourced under the MIT License. For details, please see the LICENSE file.
7+
8+
# This is a basic example of how to use DID WBA authentication.
9+
# It first creates a DID document and private keys.
10+
# Then it uploads the DID document to the server.
11+
# Then it generates an authentication header and tests the DID authentication.
12+
13+
import os
14+
import sys
15+
import json
16+
import secrets
17+
import asyncio
18+
import aiohttp
19+
import logging
20+
from pathlib import Path
21+
from cryptography.hazmat.primitives import serialization, hashes
22+
from cryptography.hazmat.primitives.asymmetric import ec
23+
from canonicaljson import encode_canonical_json
24+
25+
from agent_connect.authentication.did_wba import (
26+
create_did_wba_document,
27+
resolve_did_wba_document,
28+
generate_auth_header
29+
)
30+
from agent_connect.utils.log_base import set_log_color_level
31+
32+
_is_local_testing = False
33+
34+
# TODO: Change to your own server domain.
35+
# Or use the test domain we provide (currently using pi-unlimited.com, will later change to agent-network-protocol.com)
36+
# SERVER_DOMAIN = "agent-network-protocol.com"
37+
SERVER_DOMAIN = "pi-unlimited.com"
38+
39+
def convert_url_for_local_testing(url: str) -> str:
40+
if _is_local_testing:
41+
url = url.replace('https://', 'http://')
42+
url = url.replace(SERVER_DOMAIN, '127.0.0.1:9000')
43+
return url
44+
45+
async def upload_did_document(url: str, did_document: dict) -> bool:
46+
"""Upload DID document to server"""
47+
try:
48+
local_url = convert_url_for_local_testing(url)
49+
logging.info("Converting URL from %s to %s", url, local_url)
50+
51+
async with aiohttp.ClientSession() as session:
52+
async with session.put(
53+
local_url,
54+
json=did_document,
55+
headers={'Content-Type': 'application/json'}
56+
) as response:
57+
return response.status == 200
58+
except Exception as e:
59+
logging.error("Failed to upload DID document: %s", e)
60+
return False
61+
62+
async def download_did_document(url: str) -> dict:
63+
"""Download DID document from server"""
64+
try:
65+
local_url = convert_url_for_local_testing(url)
66+
logging.info("Converting URL from %s to %s", url, local_url)
67+
68+
async with aiohttp.ClientSession() as session:
69+
async with session.get(local_url) as response:
70+
if response.status == 200:
71+
return await response.json()
72+
logging.warning("Failed to download DID document, status: %d", response.status)
73+
return None
74+
except Exception as e:
75+
logging.error("Failed to download DID document: %s", e)
76+
return None
77+
78+
async def test_did_auth(url: str, auth_header: str) -> tuple[bool, str]:
79+
"""Test DID authentication and get token"""
80+
try:
81+
local_url = convert_url_for_local_testing(url)
82+
logging.info("Converting URL from %s to %s", url, local_url)
83+
84+
async with aiohttp.ClientSession() as session:
85+
async with session.get(
86+
local_url,
87+
headers={'Authorization': auth_header}
88+
) as response:
89+
token = response.headers.get('Authorization', '')
90+
if token.startswith('Bearer '):
91+
token = token[7:] # Remove 'Bearer ' prefix
92+
return response.status == 200, token
93+
except Exception as e:
94+
logging.error("DID authentication test failed: %s", e)
95+
return False, ''
96+
97+
def save_private_key(unique_id: str, keys: dict, did_document: dict) -> str:
98+
"""Save private keys and DID document to user directory and return the user directory path"""
99+
current_dir = Path(__file__).parent.absolute()
100+
user_dir = current_dir / "did_keys" / f"user_{unique_id}"
101+
# Create parent directories if they don't exist
102+
user_dir.mkdir(parents=True, exist_ok=True)
103+
104+
# Save private keys
105+
for method_fragment, (private_key_bytes, _) in keys.items():
106+
private_key_path = user_dir / f"{method_fragment}_private.pem"
107+
with open(private_key_path, 'wb') as f:
108+
f.write(private_key_bytes)
109+
logging.info("Saved private key '%s' to %s", method_fragment, private_key_path)
110+
111+
# Save DID document
112+
did_path = user_dir / "did.json"
113+
with open(did_path, 'w', encoding='utf-8') as f:
114+
json.dump(did_document, f, indent=2)
115+
logging.info("Saved DID document to %s", did_path)
116+
117+
return str(user_dir)
118+
119+
def load_private_key(private_key_dir: str, method_fragment: str) -> ec.EllipticCurvePrivateKey:
120+
"""Load private key from file"""
121+
key_dir = Path(private_key_dir)
122+
key_path = key_dir / f"{method_fragment}_private.pem"
123+
124+
logging.info("Loading private key from %s", key_path)
125+
with open(key_path, 'rb') as f:
126+
private_key_bytes = f.read()
127+
return serialization.load_pem_private_key(
128+
private_key_bytes,
129+
password=None
130+
)
131+
132+
def sign_callback(content: bytes, method_fragment: str) -> bytes:
133+
"""Sign content using private key"""
134+
# Load private key using the global variable
135+
private_key = load_private_key(sign_callback.private_key_dir, method_fragment)
136+
137+
# Sign the content
138+
signature = private_key.sign(
139+
content,
140+
ec.ECDSA(hashes.SHA256())
141+
)
142+
return signature
143+
144+
async def main():
145+
# 1. Generate unique identifier (8 bytes = 16 hex characters)
146+
unique_id = secrets.token_hex(8)
147+
148+
# 2. Set server information
149+
server_domain = SERVER_DOMAIN
150+
base_path = f"/wba/user/{unique_id}"
151+
did_path = f"{base_path}/did.json"
152+
153+
# 3. Create DID document
154+
logging.info("Creating DID document...")
155+
did_document, keys = create_did_wba_document(
156+
hostname=server_domain,
157+
path_segments=["wba", "user", unique_id]
158+
)
159+
160+
# 4. Save private keys, DID document and set path for sign_callback
161+
user_dir = save_private_key(unique_id, keys, did_document)
162+
sign_callback.private_key_dir = user_dir
163+
164+
# 5. Upload DID document (This should be stored on your server)
165+
document_url = f"https://{server_domain}{did_path}"
166+
logging.info("Uploading DID document to %s", document_url)
167+
success = await upload_did_document(document_url, did_document)
168+
if not success:
169+
logging.error("Failed to upload DID document")
170+
return
171+
logging.info("DID document uploaded successfully")
172+
173+
# 7. Generate authentication header
174+
logging.info("Generating authentication header...")
175+
auth_header = generate_auth_header(
176+
did_document,
177+
server_domain,
178+
sign_callback
179+
)
180+
181+
# 8. Test DID authentication and get token
182+
test_url = f"https://{server_domain}/wba/test"
183+
logging.info("Testing DID authentication at %s", test_url)
184+
auth_success, token = await test_did_auth(test_url, auth_header)
185+
186+
if not auth_success or not token:
187+
logging.error(f"DID authentication test failed. auth_success: {auth_success}, token: {token}")
188+
return
189+
190+
logging.info("DID authentication test successful")
191+
192+
if __name__ == "__main__":
193+
set_log_color_level(logging.INFO)
194+
asyncio.run(main())
195+
196+
197+
198+
199+
200+
201+

‎examples/did_wba_examples/client.py‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
# AgentConnect: https://github.com/chgaowei/AgentConnect
2+
# Author: GaoWei Chang
3+
# Email: chgaowei@gmail.com
4+
# Website: https://agent-network-protocol.com/
5+
#
6+
# This project is open-sourced under the MIT License. For details, please see the LICENSE file.
7+
8+
# This is a client example used to test whether your server supports DID WBA authentication.
9+
# It uses a pre-created DID document and private key to access a test interface on your server.
10+
# If it returns 200, it indicates that the server supports DID WBA authentication.
11+
12+
import asyncio
13+
import json
14+
import logging
15+
import aiohttp
16+
from cryptography.hazmat.primitives import serialization, hashes
17+
from cryptography.hazmat.primitives.asymmetric import ec
18+
from agent_connect.authentication.did_wba import (
19+
resolve_did_wba_document,
20+
generate_auth_header
21+
)
22+
from agent_connect.utils.log_base import set_log_color_level
23+
24+
# THIS IS A TEST DID DOCUMENT AND PRIVATE KEY
25+
CLIENT_DID = "did:wba:pi-unlimited.com:wba:user:1b299d9faa38ebaf"
26+
CLIENT_DID_DOCUMENT = '''{
27+
"@context": [
28+
"https://www.w3.org/ns/did/v1",
29+
"https://w3id.org/security/suites/jws-2020/v1",
30+
"https://w3id.org/security/suites/secp256k1-2019/v1"
31+
],
32+
"id": "did:wba:pi-unlimited.com:wba:user:1b299d9faa38ebaf",
33+
"verificationMethod": [
34+
{
35+
"id": "did:wba:pi-unlimited.com:wba:user:1b299d9faa38ebaf#key-1",
36+
"type": "EcdsaSecp256k1VerificationKey2019",
37+
"controller": "did:wba:pi-unlimited.com:wba:user:1b299d9faa38ebaf",
38+
"publicKeyJwk": {
39+
"kty": "EC",
40+
"crv": "secp256k1",
41+
"x": "P-vJSQGRoUbI2xnxywhVqblQ_hG0U0X-gja0JlfEWMg",
42+
"y": "MRTV4uHKbNvkRRBknkXjqzMfyCMGFtBw5Qlild8aLZI",
43+
"kid": "rs-lu83Tv498ETdLT_8wIC1sdFsM4ON6MKR7LQ0pI3I"
44+
}
45+
}
46+
],
47+
"authentication": [
48+
"did:wba:pi-unlimited.com:wba:user:1b299d9faa38ebaf#key-1"
49+
]
50+
}'''
51+
52+
CLIENT_PRIVATE_KEY = '''-----BEGIN PRIVATE KEY-----
53+
MIGEAgEAMBAGByqGSM49AgEGBSuBBAAKBG0wawIBAQQgumb+i8ZYrGBUB1U8HkS5
54+
Mqe4cyelFE2z7RqYYyEN4o6hRANCAAQ/68lJAZGhRsjbGfHLCFWpuVD+EbRTRf6C
55+
NrQmV8RYyDEU1eLhymzb5EUQZJ5F46szH8gjBhbQcOUJYpXfGi2S
56+
-----END PRIVATE KEY-----
57+
'''
58+
59+
# TODO: Change to your own server domain.
60+
TEST_DOMAIN = "pi-unlimited.com"
61+
62+
def load_private_key(private_key_pem: str) -> ec.EllipticCurvePrivateKey:
63+
"""Load private key from PEM string"""
64+
return serialization.load_pem_private_key(
65+
private_key_pem.encode(),
66+
password=None
67+
)
68+
69+
def sign_callback(content: bytes, method_fragment: str) -> bytes:
70+
"""Sign content using private key"""
71+
private_key = load_private_key(CLIENT_PRIVATE_KEY)
72+
signature = private_key.sign(
73+
content,
74+
ec.ECDSA(hashes.SHA256())
75+
)
76+
return signature
77+
78+
async def test_did_auth(url: str, auth_header: str) -> tuple[bool, str]:
79+
"""Test DID authentication and get token"""
80+
try:
81+
async with aiohttp.ClientSession() as session:
82+
async with session.get(
83+
url,
84+
headers={'Authorization': auth_header}
85+
) as response:
86+
token = response.headers.get('Authorization', '')
87+
if token.startswith('Bearer '):
88+
token = token[7:] # Remove 'Bearer ' prefix
89+
return response.status == 200, token
90+
except Exception as e:
91+
logging.error("DID authentication test failed: %s", e)
92+
return False, ''
93+
94+
async def main():
95+
# 1. Generate authentication header
96+
logging.info("Generating authentication header...")
97+
did_document = json.loads(CLIENT_DID_DOCUMENT)
98+
auth_header = generate_auth_header(
99+
did_document,
100+
TEST_DOMAIN,
101+
sign_callback
102+
)
103+
104+
# 2. Test DID authentication
105+
test_url = f"https://{TEST_DOMAIN}/wba/test"
106+
logging.info("Testing DID authentication at %s", test_url)
107+
auth_success, token = await test_did_auth(test_url, auth_header)
108+
109+
if auth_success:
110+
logging.info("DID authentication test successful!")
111+
logging.info(f"Received token: {token}")
112+
else:
113+
logging.error("DID authentication test failed")
114+
115+
if __name__ == "__main__":
116+
set_log_color_level(logging.INFO)
117+
asyncio.run(main())
118+
119+
120+
121+

0 commit comments

Comments
 (0)