-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathagent.py
More file actions
147 lines (121 loc) · 4.58 KB
/
Copy pathagent.py
File metadata and controls
147 lines (121 loc) · 4.58 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
"""Agent — an ephemeral per-task principal created by AgentWritApp.
Maps to the broker's agent identity model: each Agent holds a SPIFFE ID
(from POST /v1/register), a JWT access token, and lifecycle methods that
call the broker directly.
Broker endpoints used:
- POST /v1/token/renew (Agent.renew) — no body, Bearer auth
- POST /v1/token/release (Agent.release) — no body, Bearer auth, 204
- POST /v1/delegate (Agent.delegate) — body + Bearer auth
ADR SDK-006: Agent has NO validate() method. Validation is the app's
responsibility — a compromised agent cannot be trusted to validate itself.
ADR SDK-008: renew() mutates in-place. Same agent, fresh token.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from agentwrit.errors import AgentWritError
from agentwrit.models import DelegatedToken, DelegationRecord
if TYPE_CHECKING:
from agentwrit.app import AgentWritApp
class Agent:
"""An ephemeral agent registered under an AgentWritApp.
Created by AgentWritApp.create_agent(). Holds the agent JWT and
a back-reference to its parent app for transport reuse.
"""
def __init__(
self,
app: AgentWritApp,
agent_id: str,
access_token: str,
expires_in: int,
scope: list[str],
task_id: str,
orch_id: str,
) -> None:
self._app = app
self.agent_id = agent_id
self.access_token = access_token
self.expires_in = expires_in
self.scope = scope
self.task_id = task_id
self.orch_id = orch_id
self._released = False
@property
def bearer_header(self) -> dict[str, str]:
"""Returns {"Authorization": "Bearer <token>"} for HTTP requests."""
return {"Authorization": f"Bearer {self.access_token}"}
def renew(self) -> None:
"""POST /v1/token/renew — renew this agent's token in place.
The broker revokes the current JTI and issues a replacement with
the same scope, TTL, and subject. Updates access_token and
expires_in on this Agent instance. The agent_id does not change.
"""
if self._released:
raise AgentWritError("agent has been released and cannot be renewed")
response = self._app._transport.request(
"POST",
"/v1/token/renew",
headers={"Authorization": f"Bearer {self.access_token}"},
)
data = response.json()
self.access_token = data["access_token"]
self.expires_in = data["expires_in"]
def release(self) -> None:
"""POST /v1/token/release — self-revoke on task completion.
Returns None on success (broker returns 204 No Content).
After calling release(), this agent is no longer usable.
Idempotent: second call is a no-op.
"""
if self._released:
return
self._app._transport.request(
"POST",
"/v1/token/release",
headers={"Authorization": f"Bearer {self.access_token}"},
)
self._released = True
def delegate(
self,
delegate_to: str,
scope: list[str],
*,
ttl: int | None = None,
) -> DelegatedToken:
"""POST /v1/delegate — create a scope-attenuated delegation token.
delegate_to: SPIFFE ID of the target agent (must already be registered).
scope: must be a subset of this agent's scope.
ttl: delegation lifetime in seconds (broker defaults to 60 if omitted).
Max delegation depth: 5.
"""
if self._released:
raise AgentWritError("agent has been released and cannot delegate")
payload: dict[str, object] = {
"delegate_to": delegate_to,
"scope": scope,
}
if ttl is not None:
payload["ttl"] = ttl
response = self._app._transport.request(
"POST",
"/v1/delegate",
json=payload,
headers={"Authorization": f"Bearer {self.access_token}"},
)
data = response.json()
chain = [
DelegationRecord(
agent=d["agent"],
scope=d["scope"],
delegated_at=d["delegated_at"],
)
for d in data.get("delegation_chain", [])
]
return DelegatedToken(
access_token=data["access_token"],
expires_in=data["expires_in"],
delegation_chain=chain,
)
def __repr__(self) -> str:
return (
f"Agent(agent_id={self.agent_id!r}, "
f"orch_id={self.orch_id!r}, task_id={self.task_id!r})"
)