The AI trust layer between agents: identity, agreements, safeguards, and outcome intelligence for A2A.

Viewing files at commit 68f53f7907caCommit detailsView latest

apps/runtime-sidecar/src/index.ts 272 lines · 9.5 KB · typescript
1
import { createServer, type IncomingHttpHeaders } from 'node:http';
2
import { chmod, mkdir, readFile, rename, writeFile } from 'node:fs/promises';
3
import { dirname } from 'node:path';
4
import {
5
  createCipheriv,
6
  createDecipheriv,
7
  createHash,
8
  generateKeyPairSync,
9
  privateDecrypt,
10
  randomBytes,
11
} from 'node:crypto';
12
import {
13
  CONNECTOR_PROFILE_REQUEST,
14
  ConnectorAgentProfileSchema,
15
  LiveSessionActivationSchema,
16
  type LiveSessionActivation,
17
} from '../../../packages/protocol/src/index.js';
18
import {
19
  createAgentRuntimeConnector,
20
  HttpAgentRuntimeAdapter,
21
  OpenClaspClient,
22
  type RuntimeSessionStore,
23
} from '../../../packages/sdk/src/index.js';
24
25
class JsonFileRuntimeSessionStore implements RuntimeSessionStore {
26
  private loading: Promise<void> | undefined;
27
  private readonly sessions = new Map<string, LiveSessionActivation>();
28
  private writes = Promise.resolve();
29
30
  private readonly key: Buffer;
31
32
  constructor(
33
    private readonly filename: string,
34
    encryptionSecret: string,
35
  ) {
36
    this.key = createHash('sha256')
37
      .update(`openclasp-runtime-sessions:${encryptionSecret}`)
38
      .digest();
39
  }
40
41
  private async load() {
42
    this.loading ??= (async () => {
43
      try {
44
        const envelope = JSON.parse(await readFile(this.filename, 'utf8')) as {
45
          version: number;
46
          iv: string;
47
          authTag: string;
48
          ciphertext: string;
49
        };
50
        if (envelope.version !== 1) throw new Error('Unsupported runtime session file');
51
        const decipher = createDecipheriv(
52
          'aes-256-gcm',
53
          this.key,
54
          Buffer.from(envelope.iv, 'base64url'),
55
        );
56
        decipher.setAuthTag(Buffer.from(envelope.authTag, 'base64url'));
57
        const plaintext = Buffer.concat([
58
          decipher.update(Buffer.from(envelope.ciphertext, 'base64url')),
59
          decipher.final(),
60
        ]).toString('utf8');
61
        const values = JSON.parse(plaintext) as unknown[];
62
        for (const value of values) {
63
          const session = LiveSessionActivationSchema.parse(value);
64
          this.sessions.set(session.interactionId, session);
65
        }
66
      } catch (error) {
67
        if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
68
      }
69
    })();
70
    await this.loading;
71
  }
72
73
  async get(interactionId: string) {
74
    await this.load();
75
    return this.sessions.get(interactionId);
76
  }
77
78
  async put(interactionId: string, session: LiveSessionActivation) {
79
    await this.load();
80
    this.sessions.set(interactionId, LiveSessionActivationSchema.parse(session));
81
    this.writes = this.writes.then(async () => {
82
      await mkdir(dirname(this.filename), { recursive: true });
83
      const temporary = `${this.filename}.${process.pid}.tmp`;
84
      const iv = randomBytes(12);
85
      const cipher = createCipheriv('aes-256-gcm', this.key, iv);
86
      const ciphertext = Buffer.concat([
87
        cipher.update(JSON.stringify([...this.sessions.values()])),
88
        cipher.final(),
89
      ]);
90
      await writeFile(
91
        temporary,
92
        JSON.stringify({
93
          version: 1,
94
          iv: iv.toString('base64url'),
95
          authTag: cipher.getAuthTag().toString('base64url'),
96
          ciphertext: ciphertext.toString('base64url'),
97
        }),
98
        { mode: 0o600 },
99
      );
100
      await rename(temporary, this.filename);
101
    });
102
    await this.writes;
103
  }
104
}
105
106
function required(name: string) {
107
  const value = process.env[name]?.trim();
108
  if (!value) throw new Error(`${name} is required`);
109
  return value;
110
}
111
112
async function readOptionalFile(filename: string) {
113
  try {
114
    return (await readFile(filename, 'utf8')).trim() || undefined;
115
  } catch (error) {
116
    if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined;
117
    throw error;
118
  }
119
}
120
121
async function saveCredential(filename: string, credential: string) {
122
  await mkdir(dirname(filename), { recursive: true });
123
  const temporary = `${filename}.${process.pid}.tmp`;
124
  await writeFile(temporary, `${credential}\n`, { mode: 0o600 });
125
  await rename(temporary, filename);
126
  await chmod(filename, 0o600);
127
}
128
129
const wait = (milliseconds: number) =>
130
  new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
131
132
function sdkBaseUrl(value: string) {
133
  return `${value.replace(/\/$/, '').replace(/\/v0\.1$/, '')}/v0.1`;
134
}
135
136
function requestHeaders(headers: IncomingHttpHeaders) {
137
  const output = new Headers();
138
  for (const [name, value] of Object.entries(headers)) {
139
    if (Array.isArray(value)) for (const entry of value) output.append(name, entry);
140
    else if (value !== undefined) output.set(name, value);
141
  }
142
  return output;
143
}
144
145
async function readBody(request: NodeJS.ReadableStream, maximumBytes = 2_000_000) {
146
  const chunks: Buffer[] = [];
147
  let size = 0;
148
  for await (const chunk of request) {
149
    const value = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
150
    size += value.byteLength;
151
    if (size > maximumBytes) throw new Error('Request body is too large');
152
    chunks.push(value);
153
  }
154
  return Buffer.concat(chunks).toString('utf8');
155
}
156
157
const runtimeUrl = new URL(required('OPENCLASP_RUNTIME_URL'));
158
if (runtimeUrl.protocol !== 'https:') throw new Error('OPENCLASP_RUNTIME_URL must use HTTPS');
159
const agentAdapterUrl = required('AGENT_ADAPTER_URL');
160
const openClaspUrl = process.env.OPENCLASP_URL?.trim() || 'https://openclasp.dev';
161
const credentialFile =
162
  process.env.OPENCLASP_CREDENTIAL_FILE?.trim() || './data/openclasp-agent-token';
163
const adapter = new HttpAgentRuntimeAdapter(agentAdapterUrl, {
164
  ...(process.env.AGENT_ADAPTER_TOKEN?.trim()
165
    ? { bearerToken: process.env.AGENT_ADAPTER_TOKEN.trim() }
166
    : {}),
167
  timeoutMs: Number(process.env.AGENT_ADAPTER_TIMEOUT_MS ?? 30_000),
168
});
169
let accessToken =
170
  process.env.OPENCLASP_AGENT_TOKEN?.trim() || (await readOptionalFile(credentialFile));
171
if (!accessToken) {
172
  const profile = ConnectorAgentProfileSchema.parse(
173
    await adapter.describeProfile(CONNECTOR_PROFILE_REQUEST),
174
  );
175
  const { publicKey, privateKey } = generateKeyPairSync('rsa', {
176
    modulusLength: 2048,
177
    publicKeyEncoding: { type: 'spki', format: 'pem' },
178
    privateKeyEncoding: { type: 'pkcs8', format: 'pem' },
179
  });
180
  const claimClient = new OpenClaspClient(sdkBaseUrl(openClaspUrl));
181
  const claim = await claimClient.createConnectorClaim({
182
    runtimeEndpoint: runtimeUrl.toString(),
183
    credentialPublicKey: publicKey,
184
    profile,
185
  });
186
  console.log(`Approve this agent: ${claim.confirmationUrl}`);
187
  while (!accessToken) {
188
    if (Date.parse(claim.expiresAt) <= Date.now()) throw new Error('Connector claim expired');
189
    await wait(Math.max(2, claim.pollIntervalSeconds) * 1000);
190
    const result = await claimClient.pollConnectorClaim(claim.claimId, claim.claimSecret);
191
    if (result.status === 'rejected') throw new Error('Connector claim was rejected');
192
    if (result.status === 'expired') throw new Error('Connector claim expired');
193
    if (result.status === 'approved' && result.credentialCiphertext) {
194
      const credential = privateDecrypt(
195
        { key: privateKey, oaepHash: 'sha256' },
196
        Buffer.from(result.credentialCiphertext, 'base64url'),
197
      ).toString('utf8');
198
      if (!credential.startsWith('oc_at_')) throw new Error('Invalid connector credential');
199
      await saveCredential(credentialFile, credential);
200
      accessToken = credential;
201
    }
202
  }
203
}
204
const client = new OpenClaspClient(sdkBaseUrl(openClaspUrl), accessToken);
205
const bootstrap = await client.getRuntimeBootstrap();
206
const sessions = new JsonFileRuntimeSessionStore(
207
  process.env.OPENCLASP_SESSION_FILE?.trim() || './data/openclasp-sessions.json',
208
  process.env.OPENCLASP_SESSION_SECRET?.trim() || accessToken,
209
);
210
const handler = createAgentRuntimeConnector({
211
  agentId: bootstrap.agentId,
212
  a2aEndpoint: runtimeUrl.toString(),
213
  openClaspUrl,
214
  adapter,
215
  sessions,
216
});
217
218
let ready = false;
219
const server = createServer(async (request, response) => {
220
  const path = new URL(request.url ?? '/', runtimeUrl).pathname;
221
  if (request.method === 'GET' && (path === '/health' || path === '/ready')) {
222
    const status = path === '/health' || ready ? 200 : 503;
223
    response.writeHead(status, { 'content-type': 'application/json' });
224
    response.end(JSON.stringify({ status: status === 200 ? 'ready' : 'starting' }));
225
    return;
226
  }
227
  if (request.method !== 'POST' || path !== runtimeUrl.pathname) {
228
    response.writeHead(404, { 'content-type': 'application/json' });
229
    response.end(JSON.stringify({ error: 'not_found' }));
230
    return;
231
  }
232
  try {
233
    const body = await readBody(request);
234
    const result = await handler(
235
      new Request(runtimeUrl, {
236
        method: 'POST',
237
        headers: requestHeaders(request.headers),
238
        body,
239
      }),
240
    );
241
    const headers: Record<string, string> = {};
242
    result.headers.forEach((value, name) => (headers[name] = value));
243
    response.writeHead(result.status, headers);
244
    response.end(await result.text());
245
  } catch (error) {
246
    response.writeHead(500, { 'content-type': 'application/json' });
247
    response.end(
248
      JSON.stringify({ error: error instanceof Error ? error.message : 'runtime_failed' }),
249
    );
250
  }
251
});
252
253
const port = Number(process.env.PORT ?? 8080);
254
const host = process.env.HOST?.trim() || '0.0.0.0';
255
await new Promise<void>((resolve, reject) => {
256
  server.once('error', reject);
257
  server.listen(port, host, resolve);
258
});
259
260
await client.connectRuntime(runtimeUrl.toString());
261
ready = true;
262
console.log(`OpenClasp runtime connected for ${bootstrap.agentId} at ${runtimeUrl.toString()}`);
263
264
const heartbeat = setInterval(() => void client.heartbeatRuntime().catch(() => undefined), 60_000);
265
heartbeat.unref();
266
267
for (const signal of ['SIGINT', 'SIGTERM'] as const) {
268
  process.once(signal, () => {
269
    clearInterval(heartbeat);
270
    server.close(() => process.exit(0));
271
  });
272
}