`lib/a2a-client.js`

268 lines
'use strict';
/**
 * A2A (Agent2Agent) CLIENT — AbramsEgo (TK-10381, Phase A)
 *
 * Client-only. Zero dependencies — uses Node ≥18 built-in fetch.
 * Egress rules: HTTPS/443 only, 15 s timeout, no cross-host redirects,
 * matching the shape egress-sentinel treats as benign (C2-lesson 2026-07-29).
 *
 * Spec reference: a2a-protocol.org/latest/specification/ (v1.0, JSON-RPC binding)
 *
 * Activation gate: the allowlist in data/a2a-agents.json starts EMPTY.
 * Nothing is consulted until Steve adds an entry AND enables the route.
 */

const { readFileSync, appendFileSync } = require('fs');
const path = require('path');

const DATA_DIR = path.join(__dirname, '..', 'data');
const AGENTS_FILE = path.join(DATA_DIR, 'a2a-agents.json');
const CONSULTS_LOG = path.join(DATA_DIR, 'a2a-consults.jsonl');

const TIMEOUT_MS = 15_000;
const MAX_POLL_ATTEMPTS = 8;
const POLL_INTERVAL_MS = 2_000;
const CARD_CACHE_TTL_MS = 60 * 60 * 1000; // 1 h
const MAX_RESPONSE_BYTES = 1_000_000; // 1 MB — a hostile peer must not OOM us
const MAX_LOGGED_ANSWER = 20_000;     // cap what an untrusted answer writes to disk

// payload linter — fields that must never leave this box
const BANNED_PAYLOAD_PATTERNS = [
  /shopify[_-]admin[_-]token/i,
  /DATABASE_URL/i,
  /\bsecret\b.*=.*[A-Za-z0-9]{16,}/i,
  /sk_live_/,
  /pk_live_/,
  /ghp_[A-Za-z0-9]{36}/,
  /ANTHROPIC_API_KEY/i,
  /OPENAI_API_KEY/i,
  /GEMINI_API_KEY/i,
  /REPLICATE_API_TOKEN/i,
];

/** Throw if the outgoing question contains banned patterns (secrets, fleet internals). */
function lintPayload(text) {
  for (const re of BANNED_PAYLOAD_PATTERNS) {
    if (re.test(text)) throw new Error(`a2a payload blocked: contains sensitive pattern (${re})`);
  }
}

/** Enforce HTTPS/443 only. Throws on http:// or non-443 https://. */
function assertSafeUrl(urlStr) {
  let u;
  try { u = new URL(urlStr); } catch { throw new Error(`a2a: invalid URL: ${urlStr}`); }
  if (u.protocol !== 'https:') throw new Error(`a2a: only HTTPS allowed, got ${u.protocol}`);
  const port = u.port ? Number(u.port) : 443;
  if (port !== 443) throw new Error(`a2a: only port 443 allowed, got ${port}`);
  // Reject embedded credentials (https://user:pass@host) — a hostile card could
  // otherwise ship userinfo outbound while still passing the same-host check.
  if (u.username || u.password) throw new Error('a2a: URL must not contain embedded credentials (userinfo)');
}

/**
 * Read a fetch response body as JSON with a hard size cap so a hostile peer can't
 * OOM the process with a giant body. Rejects on Content-Length over the cap and
 * also truncates-and-fails if the streamed body exceeds it.
 */
async function readJsonCapped(res, maxBytes = MAX_RESPONSE_BYTES) {
  const len = Number(res.headers.get('content-length') || 0);
  if (len && len > maxBytes) throw new Error(`a2a: response too large (${len} bytes > ${maxBytes} cap)`);
  const text = await res.text();
  if (text.length > maxBytes) throw new Error(`a2a: response body exceeded ${maxBytes} byte cap`);
  return JSON.parse(text);
}

/**
 * Throw if targetUrl's host differs from baseUrl's host. A fetched Agent Card is
 * UNTRUSTED input: its `url` field could otherwise pivot the RPC call (SSRF) to
 * any other https:443 host. The RPC endpoint must stay on the allowlisted host.
 */
function assertSameHost(baseUrl, targetUrl) {
  let a, b;
  try { a = new URL(baseUrl); b = new URL(targetUrl); }
  catch { throw new Error('a2a: unparseable URL in host check'); }
  if (a.host !== b.host) {
    throw new Error(`a2a: agent card url host "${b.host}" != allowlisted host "${a.host}" (blocked card-driven redirect)`);
  }
}

/** fetch with a hard timeout and no-redirect enforcement. */
async function fetchSafe(url, opts = {}) {
  assertSafeUrl(url);
  const ctrl = new AbortController();
  const timer = setTimeout(() => ctrl.abort(), TIMEOUT_MS);
  try {
    const res = await fetch(url, {
      ...opts,
      signal: ctrl.signal,
      redirect: 'error', // never follow redirects to a different host
    });
    return res;
  } finally {
    clearTimeout(timer);
  }
}

/** In-process Agent Card cache: { [url]: { card, fetchedAt } } */
const cardCache = {};

/**
 * Fetch and validate an agent's card from /.well-known/agent-card.json.
 * Returns the parsed card object. Caches for 1 h.
 */
async function fetchAgentCard(baseUrl) {
  const cacheKey = baseUrl.replace(/\/$/, '');
  const cached = cardCache[cacheKey];
  if (cached && Date.now() - cached.fetchedAt < CARD_CACHE_TTL_MS) return cached.card;

  const cardUrl = `${cacheKey}/.well-known/agent-card.json`;
  const res = await fetchSafe(cardUrl, { headers: { Accept: 'application/json' } });
  if (!res.ok) throw new Error(`a2a: agent card fetch failed ${res.status} from ${cardUrl}`);
  const card = await readJsonCapped(res);
  if (!card || typeof card !== 'object') throw new Error('a2a: invalid agent card (not an object)');
  if (!card.name || !card.url) throw new Error('a2a: agent card missing name/url fields');
  cardCache[cacheKey] = { card, fetchedAt: Date.now() };
  return card;
}

/** Build a JSON-RPC 2.0 request body for SendMessage. */
function buildSendMessage(question, taskId) {
  return {
    jsonrpc: '2.0',
    id: taskId,
    method: 'agent/sendMessage',
    params: {
      message: {
        role: 'user',
        parts: [{ kind: 'text', text: question }],
      },
    },
  };
}

/** Build a JSON-RPC 2.0 request for GetTask. */
function buildGetTask(taskId) {
  return {
    jsonrpc: '2.0',
    id: `${taskId}-get`,
    method: 'agent/getTask',
    params: { taskId },
  };
}

/**
 * Send a JSON-RPC request to the agent's primary endpoint.
 * Reads the endpoint from the Agent Card or falls back to baseUrl + '/'.
 */
async function rpcCall(baseUrl, card, body, authHeader) {
  const rpcUrl = card?.url || `${baseUrl.replace(/\/$/, '')}/`;
  // Card is untrusted: keep the RPC on the allowlisted host (SSRF guard).
  assertSameHost(baseUrl, rpcUrl);

  const headers = { 'Content-Type': 'application/json', Accept: 'application/json' };
  if (authHeader) headers['Authorization'] = authHeader;

  const res = await fetchSafe(rpcUrl, {
    method: 'POST',
    headers,
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error(`a2a: RPC ${body.method} returned ${res.status}`);
  return readJsonCapped(res);
}

/** Pull trimmed text from an array of A2A message Parts, ignoring malformed entries. */
function collectTextParts(partsArray, out) {
  if (!Array.isArray(partsArray)) return;
  for (const p of partsArray) {
    if (p && p.kind === 'text' && typeof p.text === 'string' && p.text.trim()) out.push(p.text.trim());
  }
}

/**
 * Extract text from a completed Task's artifacts, or from an inline message result.
 * Hardened against malformed/hostile responses: non-array parts/artifacts and null
 * entries are IGNORED rather than throwing — the RPC result is untrusted input.
 */
function extractText(rpcResult) {
  const parts = [];
  // inline result (no task opened)
  collectTextParts(rpcResult?.result?.message?.parts, parts);
  // task result — each artifact carries its own parts array
  const task = rpcResult?.result?.task ?? rpcResult?.result;
  if (Array.isArray(task?.artifacts)) {
    for (const art of task.artifacts) collectTextParts(art?.parts, parts);
  }
  return parts.join('\n').trim() || null;
}

/**
 * High-level consult: send a question to a named agent (by URL) and return the answer.
 * Polls GetTask for up to MAX_POLL_ATTEMPTS if a task is opened.
 *
 * @param {string} agentUrl   base URL of the A2A server (https only, port 443)
 * @param {string} question   the question to ask
 * @param {object} opts
 * @param {string} [opts.authHeader]   optional "Bearer …" / "ApiKey …" header value
 * @param {string} [opts.agentName]    human label for logging
 * @returns {{ answer: string|null, taskId: string, pollCount: number }}
 */
async function consult(agentUrl, question, opts = {}) {
  lintPayload(question);
  const card = await fetchAgentCard(agentUrl);
  const taskId = `abramsego-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;

  let rpcResult = await rpcCall(agentUrl, card, buildSendMessage(question, taskId), opts.authHeader);
  let pollCount = 0;
  let answer = extractText(rpcResult);

  // If a task was opened, poll GetTask
  const openedTaskId = rpcResult?.result?.task?.id ?? (rpcResult?.result?.id);
  if (!answer && openedTaskId) {
    for (let i = 0; i < MAX_POLL_ATTEMPTS; i++) {
      const taskState = rpcResult?.result?.task?.status?.state ?? rpcResult?.result?.status?.state;
      if (taskState === 'completed' || taskState === 'failed' || taskState === 'canceled') break;
      await new Promise(r => setTimeout(r, POLL_INTERVAL_MS));
      rpcResult = await rpcCall(agentUrl, card, buildGetTask(openedTaskId), opts.authHeader);
      pollCount++;
      answer = extractText(rpcResult);
      if (answer) break;
    }
  }

  return { answer, taskId, pollCount };
}

/** Load the allowlist from data/a2a-agents.json. Returns [] if file missing/empty. */
function loadAllowlist() {
  try {
    const raw = readFileSync(AGENTS_FILE, 'utf8');
    const arr = JSON.parse(raw);
    return Array.isArray(arr) ? arr : [];
  } catch { return []; }
}

/** Look up a named agent entry from the allowlist. Throws if not found. */
function resolveAgent(name) {
  const list = loadAllowlist();
  const entry = list.find(a => a.name === name);
  if (!entry) throw new Error(`a2a: agent "${name}" not in allowlist (data/a2a-agents.json)`);
  return entry;
}

/** Append a consult result to the JSONL log. Fire-and-forget (errors logged, not thrown). */
function logConsult(entry) {
  try {
    // Truncate an untrusted answer before it lands on disk (unbounded-log guard).
    const safe = { ...entry, ts: new Date().toISOString() };
    if (typeof safe.answer === 'string' && safe.answer.length > MAX_LOGGED_ANSWER) {
      safe.answer = safe.answer.slice(0, MAX_LOGGED_ANSWER) + `…[truncated ${safe.answer.length - MAX_LOGGED_ANSWER} chars]`;
    }
    appendFileSync(CONSULTS_LOG, JSON.stringify(safe) + '\n');
  } catch (e) {
    console.error('a2a: consult log write failed:', e.message);
  }
}

module.exports = { fetchAgentCard, consult, loadAllowlist, resolveAgent, logConsult, lintPayload, assertSafeUrl, assertSameHost, readJsonCapped, MAX_RESPONSE_BYTES, extractText };