mirror of
https://github.com/BradGroux/veritas-kanban.git
synced 2026-08-28 02:44:59 +00:00
feat: complete governed run egress (#1080)
* feat: broker scoped egress approvals * feat: add authenticated SOCKS5 egress * feat: support operator upstream egress proxies * test: hoist workflow service mocks
This commit is contained in:
Brad Groux
GitHub
parent
e15bd320e8
commit
e69bfe8c14
18 changed files with 1229 additions and 130 deletions
|
|
@ -62,6 +62,11 @@ VERITAS_ADMIN_KEY=
|
|||
# http://127.0.0.1:5173,http://127.0.0.1:3000
|
||||
# CORS_ORIGINS=http://localhost:5173,http://localhost:3000
|
||||
|
||||
# Optional operator HTTP proxy for selective run-scoped egress.
|
||||
# Destination policy is still evaluated and the pinned IP is sent through CONNECT.
|
||||
# Credentials are held in memory and are not persisted in launch or telemetry evidence.
|
||||
# VERITAS_EGRESS_UPSTREAM_PROXY=http://proxy-user:proxy-password@proxy.internal:3128
|
||||
|
||||
# ── Logging ──────────────────────────────────────────────────
|
||||
# Pino log level: fatal | error | warn | info | debug | trace | silent
|
||||
# LOG_LEVEL=info
|
||||
|
|
|
|||
|
|
@ -817,10 +817,28 @@ the trace ID.
|
|||
Selective network policies for local task and workflow providers start an
|
||||
authenticated, run-scoped loopback gateway before dispatch. Durable launch and
|
||||
gateway evidence records only the injected proxy key names and policy digest;
|
||||
tokens and proxy URLs are never persisted. HTTP, HTTPS CONNECT, and plaintext WebSocket
|
||||
transports use the evaluated DNS address, and the gateway is stopped during
|
||||
terminal cleanup. OpenClaw is provider-managed and cannot receive this local
|
||||
boundary, so a required fine-grained policy blocks its launch.
|
||||
tokens and proxy URLs are never persisted. HTTP, HTTPS CONNECT, and plaintext
|
||||
WebSocket transports use the evaluated DNS address. `ALL_PROXY` exposes a
|
||||
separate authenticated `socks5h` listener with the same host, address, approval,
|
||||
and audit policy; encrypted SOCKS tunnels fail closed when method or path
|
||||
inspection would be required. Both listeners stop during terminal cleanup.
|
||||
OpenClaw is provider-managed and cannot receive this local boundary, so a
|
||||
required fine-grained policy blocks its launch.
|
||||
|
||||
Operators that require a corporate or audited proxy can set
|
||||
`VERITAS_EGRESS_UPSTREAM_PROXY` to an HTTP proxy origin. The gateway evaluates
|
||||
the destination first, pins the resolved address, and then opens an upstream
|
||||
CONNECT tunnel to that address. Proxy credentials remain memory-only; durable
|
||||
evidence records only `upstreamMode: http-connect`.
|
||||
|
||||
When a preset enables scoped approvals, an otherwise eligible block creates a
|
||||
durable, exact-action network approval and holds the request at the gateway.
|
||||
Only an approved request proceeds. Explicit host denies and protected address
|
||||
classes are never approval-eligible. Gateway shutdown aborts pending waits.
|
||||
Every final decision also emits a `network.egress` telemetry record containing
|
||||
the hashed host key, run key, policy digest, protocol, port, decision, reason,
|
||||
and approval ID when present. URLs, query strings, headers, bodies, proxy
|
||||
credentials, and raw paths are excluded from telemetry.
|
||||
|
||||
Local ACP, Claude Code, Codex app-server, Codex CLI, and Hermes processes use a
|
||||
version-bound `codex sandbox` wrapper when its credential-free conformance
|
||||
|
|
|
|||
|
|
@ -1144,6 +1144,11 @@ GET /api/telemetry/count # Event counts
|
|||
GET /api/telemetry/export # Export events (CSV/JSON)
|
||||
```
|
||||
|
||||
The run-scoped gateway emits `network.egress` events internally. These records
|
||||
contain run and gateway keys, policy digest, hashed host key, protocol, port,
|
||||
decision, reason, and optional approval correlation. They never contain full
|
||||
URLs, query strings, headers, bodies, proxy credentials, or raw paths.
|
||||
|
||||
---
|
||||
|
||||
## Health
|
||||
|
|
|
|||
|
|
@ -536,8 +536,9 @@ All variables are set in `server/.env` (or passed as environment variables in Do
|
|||
### Networking & Security
|
||||
|
||||
| Variable | Default | Description |
|
||||
| ----------------- | ------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| ------------------------------- | ------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `TRUST_PROXY` | — | Express trust proxy setting for reverse proxy deployments. Common: `1` (single hop), `loopback`. Required for correct rate limiting behind nginx/Caddy/Traefik. `true` is blocked for safety |
|
||||
| `VERITAS_EGRESS_UPSTREAM_PROXY` | — | Optional operator HTTP proxy for policy-approved run egress. The gateway tunnels to the DNS-pinned destination and never persists proxy credentials |
|
||||
| `CORS_ORIGINS` | `http://localhost:3000,http://localhost:5173,...` | Comma-separated list of allowed CORS origins |
|
||||
| `RATE_LIMIT_MAX` | `300` | Max API requests per minute per IP (localhost exempt). Auth endpoints have a stricter 15 req/min limit |
|
||||
| `CSP_REPORT_ONLY` | `false` | Use Content-Security-Policy-Report-Only instead of enforcing |
|
||||
|
|
|
|||
|
|
@ -1177,10 +1177,18 @@ Reusable launch-time sandbox presets for provider execution guardrails.
|
|||
rules take precedence, unsafe global allow wildcards fail validation, and a
|
||||
deterministic policy digest binds later gateway launch evidence.
|
||||
- Selective local-provider policies start an authenticated loopback gateway
|
||||
before provider dispatch. Veritas injects HTTP, HTTPS, and all-proxy variables,
|
||||
clears proxy bypass variables, pins the evaluated DNS address for transport,
|
||||
and stops the gateway with the run. Remote OpenClaw execution fails closed
|
||||
when the preset requires this local gateway.
|
||||
before provider dispatch. Veritas injects HTTP and HTTPS proxy variables plus
|
||||
an authenticated `socks5h` all-proxy listener, clears proxy bypass variables,
|
||||
pins the evaluated DNS address for transport, and stops both listeners with
|
||||
the run. Remote OpenClaw execution fails closed when the preset requires this
|
||||
local gateway.
|
||||
- Optional `VERITAS_EGRESS_UPSTREAM_PROXY` routing sends only policy-approved,
|
||||
DNS-pinned destinations through an operator HTTP CONNECT proxy. Credentials
|
||||
stay memory-only and evidence exposes only the upstream mode.
|
||||
- Approval-eligible blocks pause at the gateway on a durable exact-action
|
||||
approval. Explicit denies and protected address classes cannot be overridden.
|
||||
Approved requests retain the approval ID in metadata-only governance and
|
||||
`network.egress` telemetry evidence.
|
||||
|
||||
**Enforcement:**
|
||||
|
||||
|
|
@ -1746,7 +1754,7 @@ New endpoints for advanced metrics and visualization (v1.6):
|
|||
|
||||
Event-based telemetry system powering dashboard analytics.
|
||||
|
||||
- **Event types** — `run.started`, `run.completed`, `run.tokens` for tracking agent execution lifecycle
|
||||
- **Event types** — `run.started`, `run.completed`, `run.tokens`, and metadata-only `network.egress` decisions for tracking execution and network policy outcomes
|
||||
- **Token tracking** — Input tokens, output tokens, cache tokens, and cost per run
|
||||
- **Duration tracking** — Millisecond-precision run duration with 7-day cap validation (604,800,000 ms)
|
||||
- **Retention policy** — Configurable retention period (default: 30 days) with automatic cleanup of old events
|
||||
|
|
|
|||
|
|
@ -63,6 +63,11 @@ CORS_ORIGINS=http://localhost:3000,http://localhost:5173,http://127.0.0.1:3000,h
|
|||
# Reverse proxy support (set when behind nginx, Caddy, Traefik, etc.)
|
||||
# TRUST_PROXY=1
|
||||
|
||||
# Optional operator HTTP proxy for selective run-scoped egress.
|
||||
# The gateway tunnels only policy-approved, DNS-pinned destinations through it.
|
||||
# Credentials remain memory-only and are excluded from durable evidence.
|
||||
# VERITAS_EGRESS_UPSTREAM_PROXY=http://proxy-user:proxy-password@proxy.internal:3128
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# DATA & STORAGE
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
|
|
|||
|
|
@ -23,12 +23,21 @@ import { workflowAdmissionStub } from './helpers/workflow-admission-stub.js';
|
|||
// Using module-level fns so vi.mock factory can close over them.
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
|
||||
const mockLoadWorkflow = vi.fn();
|
||||
const mockPrepareStep = vi.fn(async (step: unknown) => ({ kind: 'non-agent', step }));
|
||||
const mockApplyPreparation = vi.fn();
|
||||
const mockExecuteStep = vi.fn();
|
||||
const mockBroadcastWorkflowStatus = vi.fn();
|
||||
const mockGetTask = vi.fn();
|
||||
const {
|
||||
mockLoadWorkflow,
|
||||
mockPrepareStep,
|
||||
mockApplyPreparation,
|
||||
mockExecuteStep,
|
||||
mockBroadcastWorkflowStatus,
|
||||
mockGetTask,
|
||||
} = vi.hoisted(() => ({
|
||||
mockLoadWorkflow: vi.fn(),
|
||||
mockPrepareStep: vi.fn(async (step: unknown) => ({ kind: 'non-agent', step })),
|
||||
mockApplyPreparation: vi.fn(),
|
||||
mockExecuteStep: vi.fn(),
|
||||
mockBroadcastWorkflowStatus: vi.fn(),
|
||||
mockGetTask: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock('../services/workflow-service.js', () => ({
|
||||
getWorkflowService: () => ({
|
||||
|
|
|
|||
|
|
@ -5,6 +5,8 @@ import type { AddressInfo } from 'node:net';
|
|||
import { EgressPolicyService } from '../services/egress-policy-service.js';
|
||||
import {
|
||||
RunEgressGatewayService,
|
||||
type RunEgressGatewayApprovalRequest,
|
||||
type RunEgressGatewayApprovalResult,
|
||||
type RunEgressGatewayHandle,
|
||||
} from '../services/run-egress-gateway-service.js';
|
||||
|
||||
|
|
@ -80,10 +82,10 @@ describe('RunEgressGatewayService', () => {
|
|||
gatewayId: gateway.gatewayId,
|
||||
runKey: expect.stringMatching(/^sha256:[a-f0-9]{64}$/),
|
||||
attributionKey: expect.stringMatching(/^sha256:[a-f0-9]{64}$/),
|
||||
protocols: ['http', 'connect', 'ws'],
|
||||
protocols: ['http', 'connect', 'ws', 'socks5'],
|
||||
});
|
||||
expect(JSON.stringify(gateway.evidence)).not.toContain(proxy.password);
|
||||
expect(gateway.environment.ALL_PROXY).toBe(gateway.environment.HTTP_PROXY);
|
||||
expect(gateway.environment.ALL_PROXY).toMatch(/^socks5h:\/\/veritas:/);
|
||||
expect(gateway.environment.NO_PROXY).toBe('');
|
||||
|
||||
await expect(gateway.stop()).resolves.toMatchObject({
|
||||
|
|
@ -132,34 +134,192 @@ describe('RunEgressGatewayService', () => {
|
|||
);
|
||||
expect(response).toContain('101 Switching Protocols');
|
||||
});
|
||||
|
||||
it('opens an authenticated SOCKS5 tunnel with the same pinned policy decision', async () => {
|
||||
const echo = net.createServer((socket) => socket.pipe(socket));
|
||||
const echoPort = await listen(echo);
|
||||
const audit = vi.fn();
|
||||
const gateway = await startGateway({
|
||||
runId: 'run-egress-socks',
|
||||
onDecision: audit,
|
||||
});
|
||||
|
||||
const tunnel = await connectSocksTunnel(gateway, '127.0.0.1', echoPort);
|
||||
await expect(roundTrip(tunnel, 'through-socks')).resolves.toBe('through-socks');
|
||||
tunnel.destroy();
|
||||
expect(audit).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
decision: expect.objectContaining({
|
||||
protocol: 'socks',
|
||||
decision: 'allow',
|
||||
}),
|
||||
})
|
||||
);
|
||||
|
||||
const restricted = await startGateway({
|
||||
runId: 'run-egress-socks-restricted',
|
||||
allowedMethods: ['GET'],
|
||||
});
|
||||
await expect(connectSocksTunnel(restricted, '127.0.0.1', echoPort)).rejects.toMatchObject({
|
||||
replyCode: 0x02,
|
||||
});
|
||||
await expect(
|
||||
connectSocksTunnel(gateway, '127.0.0.1', echoPort, 'invalid-token')
|
||||
).rejects.toMatchObject({
|
||||
authStatus: 0x01,
|
||||
});
|
||||
});
|
||||
|
||||
it('uses an operator upstream proxy without exposing its credentials in evidence', async () => {
|
||||
const destination = http.createServer((_request, response) => response.end('via-upstream'));
|
||||
const destinationPort = await listen(destination);
|
||||
const observed: Array<{ authority: string; authorization?: string }> = [];
|
||||
const upstream = http.createServer();
|
||||
upstream.on('connect', (request, clientSocket, head) => {
|
||||
observed.push({
|
||||
authority: request.url ?? '',
|
||||
authorization:
|
||||
typeof request.headers['proxy-authorization'] === 'string'
|
||||
? request.headers['proxy-authorization']
|
||||
: undefined,
|
||||
});
|
||||
const authority = new URL(`http://${request.url}`);
|
||||
const destinationSocket = net.connect(Number(authority.port), authority.hostname);
|
||||
destinationSocket.once('connect', () => {
|
||||
clientSocket.write('HTTP/1.1 200 Connection Established\r\n\r\n');
|
||||
if (head.length > 0) destinationSocket.write(head);
|
||||
clientSocket.pipe(destinationSocket);
|
||||
destinationSocket.pipe(clientSocket);
|
||||
});
|
||||
destinationSocket.once('error', () => clientSocket.destroy());
|
||||
});
|
||||
const upstreamPort = await listen(upstream);
|
||||
const gateway = await startGateway({
|
||||
runId: 'run-egress-upstream',
|
||||
upstreamProxyUrl: `http://proxy-user:proxy-pass@127.0.0.1:${upstreamPort}`,
|
||||
});
|
||||
|
||||
await expect(
|
||||
proxyRequest(gateway, `http://127.0.0.1:${destinationPort}/`, 'GET')
|
||||
).resolves.toMatchObject({
|
||||
statusCode: 200,
|
||||
body: 'via-upstream',
|
||||
});
|
||||
expect(observed).toEqual([
|
||||
{
|
||||
authority: `127.0.0.1:${destinationPort}`,
|
||||
authorization: `Basic ${Buffer.from('proxy-user:proxy-pass').toString('base64')}`,
|
||||
},
|
||||
]);
|
||||
expect(gateway.evidence.upstreamMode).toBe('http-connect');
|
||||
expect(JSON.stringify(gateway.evidence)).not.toContain('proxy-pass');
|
||||
});
|
||||
|
||||
it('pauses approval-eligible requests and never lets approval override an explicit deny', async () => {
|
||||
const upstream = http.createServer((_request, response) => response.end('approved-egress'));
|
||||
const upstreamPort = await listen(upstream);
|
||||
const onApprovalRequired = vi.fn(async (): Promise<RunEgressGatewayApprovalResult> => ({
|
||||
approvalId: 'runapproval_network',
|
||||
approved: true,
|
||||
}));
|
||||
const audit = vi.fn();
|
||||
const approvedGateway = await startGateway({
|
||||
runId: 'run-egress-approved',
|
||||
defaultEgress: 'deny',
|
||||
allowedHosts: [],
|
||||
allowApprovals: true,
|
||||
onApprovalRequired,
|
||||
onDecision: audit,
|
||||
});
|
||||
|
||||
await expect(
|
||||
proxyRequest(
|
||||
approvedGateway,
|
||||
`http://127.0.0.1:${upstreamPort}/approved?secret=not-audited`,
|
||||
'GET'
|
||||
)
|
||||
).resolves.toMatchObject({
|
||||
statusCode: 200,
|
||||
body: 'approved-egress',
|
||||
});
|
||||
expect(onApprovalRequired).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
host: '127.0.0.1',
|
||||
port: upstreamPort,
|
||||
path: '/approved',
|
||||
decision: expect.objectContaining({
|
||||
decision: 'block',
|
||||
reason: 'default-deny',
|
||||
approvalEligible: true,
|
||||
}),
|
||||
})
|
||||
);
|
||||
expect(audit).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
decision: expect.objectContaining({
|
||||
decision: 'allow',
|
||||
reason: 'allowed-by-approval',
|
||||
policyReason: 'default-deny',
|
||||
approvalId: 'runapproval_network',
|
||||
}),
|
||||
})
|
||||
);
|
||||
expect(JSON.stringify(audit.mock.calls)).not.toContain('not-audited');
|
||||
|
||||
const deniedApproval = vi.fn();
|
||||
const deniedGateway = await startGateway({
|
||||
runId: 'run-egress-denied',
|
||||
defaultEgress: 'allow',
|
||||
deniedHosts: ['127.0.0.1'],
|
||||
allowApprovals: true,
|
||||
onApprovalRequired: deniedApproval,
|
||||
});
|
||||
await expect(
|
||||
proxyRequest(deniedGateway, `http://127.0.0.1:${upstreamPort}/`, 'GET')
|
||||
).resolves.toMatchObject({
|
||||
statusCode: 403,
|
||||
body: expect.stringContaining('denied-host'),
|
||||
});
|
||||
expect(deniedApproval).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
async function startGateway(
|
||||
overrides: {
|
||||
runId?: string;
|
||||
defaultEgress?: 'allow' | 'deny';
|
||||
allowedHosts?: string[];
|
||||
deniedHosts?: string[];
|
||||
allowedMethods?: string[];
|
||||
allowedPathPrefixes?: string[];
|
||||
allowApprovals?: boolean;
|
||||
upstreamProxyUrl?: string;
|
||||
onDecision?: (event: unknown) => void;
|
||||
onApprovalRequired?: (
|
||||
request: RunEgressGatewayApprovalRequest
|
||||
) => Promise<RunEgressGatewayApprovalResult>;
|
||||
} = {}
|
||||
): Promise<RunEgressGatewayHandle> {
|
||||
const service = new RunEgressGatewayService(policyService);
|
||||
const gateway = await service.start({
|
||||
runId: overrides.runId ?? `run-egress-${openGateways.length}`,
|
||||
policy: policyService.compile({
|
||||
defaultEgress: 'deny',
|
||||
allowedHosts: ['127.0.0.1'],
|
||||
deniedHosts: [],
|
||||
defaultEgress: overrides.defaultEgress ?? 'deny',
|
||||
allowedHosts: overrides.allowedHosts ?? ['127.0.0.1'],
|
||||
deniedHosts: overrides.deniedHosts ?? [],
|
||||
allowedMethods: overrides.allowedMethods ?? [],
|
||||
allowedPathPrefixes: overrides.allowedPathPrefixes ?? [],
|
||||
blockPrivateNetwork: false,
|
||||
blockMetadataEndpoints: false,
|
||||
blockLoopback: false,
|
||||
allowApprovals: false,
|
||||
allowApprovals: overrides.allowApprovals ?? false,
|
||||
dangerouslyAllowGlobalWildcard: false,
|
||||
}),
|
||||
requestTimeoutMs: 5_000,
|
||||
idleTimeoutMs: 5_000,
|
||||
upstreamProxyUrl: overrides.upstreamProxyUrl,
|
||||
onDecision: overrides.onDecision,
|
||||
onApprovalRequired: overrides.onApprovalRequired,
|
||||
});
|
||||
openGateways.push(gateway);
|
||||
return gateway;
|
||||
|
|
@ -272,6 +432,56 @@ function roundTrip(socket: Socket, value: string): Promise<string> {
|
|||
});
|
||||
}
|
||||
|
||||
async function connectSocksTunnel(
|
||||
gateway: RunEgressGatewayHandle,
|
||||
host: string,
|
||||
port: number,
|
||||
passwordOverride?: string
|
||||
): Promise<Socket> {
|
||||
const proxy = new URL(gateway.environment.ALL_PROXY);
|
||||
const socket = net.connect(Number(proxy.port), proxy.hostname);
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
socket.once('connect', resolve);
|
||||
socket.once('error', reject);
|
||||
});
|
||||
socket.write(Buffer.from([0x05, 0x01, 0x02]));
|
||||
expect(await readSocketBytes(socket, 2)).toEqual(Buffer.from([0x05, 0x02]));
|
||||
|
||||
const username = Buffer.from(proxy.username);
|
||||
const password = Buffer.from(passwordOverride ?? proxy.password);
|
||||
socket.write(
|
||||
Buffer.concat([
|
||||
Buffer.from([0x01, username.length]),
|
||||
username,
|
||||
Buffer.from([password.length]),
|
||||
password,
|
||||
])
|
||||
);
|
||||
const auth = await readSocketBytes(socket, 2);
|
||||
if (auth[1] !== 0x00) {
|
||||
socket.destroy();
|
||||
throw { authStatus: auth[1] };
|
||||
}
|
||||
|
||||
const address = Buffer.from(host.split('.').map(Number));
|
||||
const portBytes = Buffer.alloc(2);
|
||||
portBytes.writeUInt16BE(port);
|
||||
socket.write(Buffer.concat([Buffer.from([0x05, 0x01, 0x00, 0x01]), address, portBytes]));
|
||||
const reply = await readSocketBytes(socket, 10);
|
||||
if (reply[1] !== 0x00) {
|
||||
socket.destroy();
|
||||
throw { replyCode: reply[1] };
|
||||
}
|
||||
return socket;
|
||||
}
|
||||
|
||||
function readSocketBytes(socket: Socket, size: number): Promise<Buffer> {
|
||||
return new Promise((resolve, reject) => {
|
||||
socket.once('data', (chunk) => resolve(Buffer.from(chunk).subarray(0, size)));
|
||||
socket.once('error', reject);
|
||||
});
|
||||
}
|
||||
|
||||
function websocketUpgrade(gateway: RunEgressGatewayHandle, target: string): Promise<string> {
|
||||
const proxy = new URL(gateway.environment.HTTP_PROXY);
|
||||
return new Promise((resolve, reject) => {
|
||||
|
|
|
|||
|
|
@ -2,7 +2,10 @@ import { describe, expect, it } from 'vitest';
|
|||
import { PROVIDER_RUNTIME_PROBE_REVISION } from '@veritas-kanban/shared';
|
||||
import { EgressPolicyService } from '../services/egress-policy-service.js';
|
||||
import { getProviderRuntimeAdapterDefinition } from '../services/provider-runtime-adapter-registry.js';
|
||||
import { runEgressPolicyRequiresGateway } from '../services/run-egress-gateway-service.js';
|
||||
import {
|
||||
RunEgressGatewayService,
|
||||
runEgressPolicyRequiresGateway,
|
||||
} from '../services/run-egress-gateway-service.js';
|
||||
|
||||
const policyService = new EgressPolicyService();
|
||||
|
||||
|
|
@ -94,4 +97,23 @@ describe('run egress launch policy', () => {
|
|||
state: 'advisory',
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects unsupported upstream proxy schemes before binding listeners', async () => {
|
||||
const gateway = new RunEgressGatewayService(policyService);
|
||||
await expect(
|
||||
gateway.start({
|
||||
runId: 'run-invalid-upstream',
|
||||
policy: policyService.compile({
|
||||
defaultEgress: 'deny',
|
||||
allowedHosts: ['api.example.com'],
|
||||
allowedMethods: [],
|
||||
allowedPathPrefixes: [],
|
||||
blockPrivateNetwork: true,
|
||||
blockMetadataEndpoints: true,
|
||||
blockLoopback: true,
|
||||
}),
|
||||
upstreamProxyUrl: 'https://proxy.example.com:8443',
|
||||
})
|
||||
).rejects.toThrow(/must be an HTTP origin/i);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -187,6 +187,7 @@ describe('Common Schemas', () => {
|
|||
'run.completed',
|
||||
'run.error',
|
||||
'run.tokens',
|
||||
'network.egress',
|
||||
'admission.tree_control',
|
||||
];
|
||||
for (const type of validTypes) {
|
||||
|
|
|
|||
|
|
@ -505,12 +505,15 @@ describe('WorkflowStepExecutor Codex integration', () => {
|
|||
VERITAS_ADMIN_KEY: process.env.VERITAS_ADMIN_KEY,
|
||||
OPENAI_API_KEY: process.env.OPENAI_API_KEY,
|
||||
VK_API_URL: process.env.VK_API_URL,
|
||||
VERITAS_EGRESS_UPSTREAM_PROXY: process.env.VERITAS_EGRESS_UPSTREAM_PROXY,
|
||||
};
|
||||
process.env.GITHUB_TOKEN = 'test-github-token';
|
||||
process.env.DATABASE_URL = 'postgres://test-secret';
|
||||
process.env.VERITAS_ADMIN_KEY = 'test-admin-key';
|
||||
process.env.OPENAI_API_KEY = 'test-openai-key';
|
||||
process.env.VK_API_URL = 'http://127.0.0.1:3001';
|
||||
process.env.VERITAS_EGRESS_UPSTREAM_PROXY =
|
||||
'http://proxy-user:proxy-password@proxy.internal:3128';
|
||||
|
||||
try {
|
||||
const executor = new WorkflowStepExecutor(tmpDir, { runtimeManifestResolver });
|
||||
|
|
@ -563,6 +566,7 @@ describe('WorkflowStepExecutor Codex integration', () => {
|
|||
expect(env?.GITHUB_TOKEN).toBeUndefined();
|
||||
expect(env?.DATABASE_URL).toBeUndefined();
|
||||
expect(env?.VERITAS_ADMIN_KEY).toBeUndefined();
|
||||
expect(env?.VERITAS_EGRESS_UPSTREAM_PROXY).toBeUndefined();
|
||||
} finally {
|
||||
for (const [key, value] of Object.entries(originalEnv)) {
|
||||
if (value === undefined) {
|
||||
|
|
|
|||
|
|
@ -81,6 +81,7 @@ export const TelemetryEventTypeSchema = z.enum([
|
|||
'run.completed',
|
||||
'run.error',
|
||||
'run.tokens',
|
||||
'network.egress',
|
||||
'admission.tree_control',
|
||||
]);
|
||||
|
||||
|
|
|
|||
|
|
@ -25,7 +25,10 @@ import { getSandboxPolicyService, type SandboxPolicyService } from './sandbox-po
|
|||
import {
|
||||
getRunEgressGatewayService,
|
||||
RUN_EGRESS_PROXY_ENVIRONMENT_KEYS,
|
||||
RUN_EGRESS_UPSTREAM_PROXY_ENV_KEY,
|
||||
runEgressPolicyRequiresGateway,
|
||||
type RunEgressGatewayApprovalRequest,
|
||||
type RunEgressGatewayApprovalResult,
|
||||
type RunEgressGatewayHandle,
|
||||
type RunEgressGatewayService,
|
||||
} from './run-egress-gateway-service.js';
|
||||
|
|
@ -78,6 +81,7 @@ import type {
|
|||
RunStartedEvent,
|
||||
RunCompletedEvent,
|
||||
RunErrorEvent,
|
||||
NetworkEgressTelemetryEvent,
|
||||
TokenTelemetryEvent,
|
||||
TaskReadinessSummary,
|
||||
SandboxPolicyDryRunResult,
|
||||
|
|
@ -258,6 +262,7 @@ import {
|
|||
} from './run-tool-bridge-service.js';
|
||||
import { getToolPolicyService } from './tool-policy-service.js';
|
||||
import { RunRecoveryPolicyService } from './run-recovery-policy-service.js';
|
||||
import { digestRunLaunchValue } from '../utils/run-launch-manifest-digest.js';
|
||||
import {
|
||||
FilesystemSandboxService,
|
||||
getFilesystemSandboxService,
|
||||
|
|
@ -2966,7 +2971,17 @@ export class ClawdbotAgentService {
|
|||
pending.egressGateway = await this.runEgressGateway.start({
|
||||
runId: attemptId,
|
||||
policy: egressPolicy,
|
||||
onDecision: (event) => {
|
||||
upstreamProxyUrl: process.env[RUN_EGRESS_UPSTREAM_PROXY_ENV_KEY],
|
||||
onApprovalRequired: (request) =>
|
||||
this.resolveRunEgressApproval(
|
||||
task,
|
||||
attemptId,
|
||||
provider,
|
||||
agent,
|
||||
runLaunchManifest,
|
||||
request
|
||||
),
|
||||
onDecision: async (event) => {
|
||||
this.recordTraceStep(attemptId, 'execute', {
|
||||
eventType: 'network.egress.decision',
|
||||
gatewayId: event.gatewayId,
|
||||
|
|
@ -2981,6 +2996,29 @@ export class ClawdbotAgentService {
|
|||
reason: event.decision.reason,
|
||||
blockedAddressClass: event.decision.blockedAddressClass,
|
||||
approvalEligible: event.decision.approvalEligible,
|
||||
approvalId: event.decision.approvalId,
|
||||
policyReason: event.decision.policyReason,
|
||||
});
|
||||
await telemetry.emit<NetworkEgressTelemetryEvent>({
|
||||
type: 'network.egress',
|
||||
taskId,
|
||||
attemptId,
|
||||
agent,
|
||||
provider,
|
||||
project: task.project,
|
||||
gatewayId: event.gatewayId,
|
||||
runKey: event.runKey,
|
||||
policyHash: event.decision.policyHash,
|
||||
protocol: event.decision.protocol,
|
||||
hostKey: event.decision.hostKey,
|
||||
port: event.decision.port,
|
||||
method: event.decision.method,
|
||||
decision: event.decision.decision,
|
||||
reason: event.decision.reason,
|
||||
policyReason: event.decision.policyReason,
|
||||
blockedAddressClass: event.decision.blockedAddressClass,
|
||||
approvalEligible: event.decision.approvalEligible,
|
||||
approvalId: event.decision.approvalId,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
|
@ -5884,6 +5922,88 @@ export class ClawdbotAgentService {
|
|||
return updateType === 'agent_message_chunk' ? summary : undefined;
|
||||
}
|
||||
|
||||
private async resolveRunEgressApproval(
|
||||
task: Task,
|
||||
attemptId: string,
|
||||
provider: ExecutableAgentProvider,
|
||||
agentId: string,
|
||||
runLaunchManifest: RunLaunchManifest,
|
||||
request: RunEgressGatewayApprovalRequest
|
||||
): Promise<RunEgressGatewayApprovalResult> {
|
||||
const phase = await this.bindPhaseApproval(task.id, attemptId, runLaunchManifest, [
|
||||
{
|
||||
dimension: 'network.egress',
|
||||
requestedScopes: [request.host],
|
||||
},
|
||||
]);
|
||||
const providerRequestId = `egress:${digestRunLaunchValue({
|
||||
gatewayId: request.gatewayId,
|
||||
protocol: request.protocol,
|
||||
host: request.host,
|
||||
port: request.port,
|
||||
method: request.method,
|
||||
path: request.path,
|
||||
policyHash: request.decision.policyHash,
|
||||
}).slice('sha256:'.length, 'sha256:'.length + 32)}`;
|
||||
const approval = await this.approvalBroker.request({
|
||||
workspaceId: 'local',
|
||||
taskId: task.id,
|
||||
attemptId,
|
||||
provider,
|
||||
agentId,
|
||||
providerRequestId,
|
||||
requestKind: 'approval',
|
||||
actionClass: 'network',
|
||||
action: `Allow ${request.protocol} egress to ${request.host}:${request.port}`,
|
||||
details: `Blocked by ${request.decision.reason}; method ${request.method ?? 'not available'}.`,
|
||||
resourceScope: [request.host, `${request.protocol}:${request.port}`],
|
||||
workingDirectory: task.git?.worktreePath,
|
||||
riskClass: 'high',
|
||||
policyReason: request.decision.reason,
|
||||
evidenceRevision: runLaunchManifest.digest,
|
||||
mobileSafe: false,
|
||||
exactAction: {
|
||||
protocol: request.protocol,
|
||||
host: request.host,
|
||||
port: request.port,
|
||||
method: request.method,
|
||||
path: request.path,
|
||||
policyHash: request.decision.policyHash,
|
||||
blockedReason: request.decision.reason,
|
||||
},
|
||||
...(phase ? { phase } : {}),
|
||||
});
|
||||
if (request.signal.aborted) {
|
||||
await this.approvalBroker.cancelAttempt(
|
||||
'local',
|
||||
task.id,
|
||||
attemptId,
|
||||
'Run egress gateway stopped.'
|
||||
);
|
||||
return { approvalId: approval.id, approved: false };
|
||||
}
|
||||
try {
|
||||
const resolved = await this.approvalBroker.awaitDecision(approval.id, {
|
||||
signal: request.signal,
|
||||
});
|
||||
return {
|
||||
approvalId: approval.id,
|
||||
approved: resolved.request.status === 'approved',
|
||||
};
|
||||
} catch (error) {
|
||||
if (request.signal.aborted) {
|
||||
await this.approvalBroker.cancelAttempt(
|
||||
'local',
|
||||
task.id,
|
||||
attemptId,
|
||||
'Run egress gateway stopped.'
|
||||
);
|
||||
return { approvalId: approval.id, approved: false };
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async resolveAcpPermission(
|
||||
task: Task,
|
||||
attemptId: string,
|
||||
|
|
@ -10061,7 +10181,10 @@ export class ClawdbotAgentService {
|
|||
Object.entries({
|
||||
...environment,
|
||||
...pending.egressGateway?.environment,
|
||||
}).filter((entry): entry is [string, string] => typeof entry[1] === 'string')
|
||||
}).filter(
|
||||
(entry): entry is [string, string] =>
|
||||
entry[0] !== RUN_EGRESS_UPSTREAM_PROXY_ENV_KEY && typeof entry[1] === 'string'
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -590,6 +590,8 @@ function telemetryTitle(event: AnyTelemetryEvent): string {
|
|||
return `Agent run error from ${event.agent}`;
|
||||
case 'run.tokens':
|
||||
return `Token usage recorded for ${event.agent}`;
|
||||
case 'network.egress':
|
||||
return `Network egress ${event.decision} for ${event.agent}`;
|
||||
case 'admission.tree_control':
|
||||
return `Execution tree ${event.action}`;
|
||||
case 'task.status_changed':
|
||||
|
|
@ -609,6 +611,9 @@ function telemetryDetail(event: AnyTelemetryEvent): string | undefined {
|
|||
if (event.type === 'run.tokens') {
|
||||
return `${(event.totalTokens ?? event.inputTokens + event.outputTokens).toLocaleString()} tokens`;
|
||||
}
|
||||
if (event.type === 'network.egress') {
|
||||
return `${event.protocol} ${event.decision}: ${event.reason}`;
|
||||
}
|
||||
if (event.type === 'task.status_changed') {
|
||||
return `${event.previousStatus ?? 'unknown'} -> ${event.status ?? 'unknown'}`;
|
||||
}
|
||||
|
|
@ -636,6 +641,16 @@ function telemetryMetadata(
|
|||
metadata.rootObjectiveKey = event.rootObjectiveKey;
|
||||
metadata.unresolvedAttempts = event.unresolvedAttempts ?? null;
|
||||
}
|
||||
if (event.type === 'network.egress') {
|
||||
metadata.gatewayId = event.gatewayId;
|
||||
metadata.runKey = event.runKey;
|
||||
metadata.policyHash = event.policyHash;
|
||||
metadata.hostKey = event.hostKey;
|
||||
metadata.port = event.port;
|
||||
metadata.decision = event.decision;
|
||||
metadata.reason = event.reason;
|
||||
metadata.approvalId = event.approvalId ?? null;
|
||||
}
|
||||
return metadata;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import http, {
|
|||
type IncomingMessage,
|
||||
type ServerResponse,
|
||||
} from 'node:http';
|
||||
import net, { type Socket } from 'node:net';
|
||||
import net, { type Server as NetServer, type Socket } from 'node:net';
|
||||
import { URL } from 'node:url';
|
||||
import {
|
||||
RUN_EGRESS_GATEWAY_EVIDENCE_SCHEMA_VERSION,
|
||||
|
|
@ -19,6 +19,7 @@ const LOOPBACK_HOST = '127.0.0.1';
|
|||
const DEFAULT_REQUEST_TIMEOUT_MS = 30_000;
|
||||
const DEFAULT_IDLE_TIMEOUT_MS = 60_000;
|
||||
const DEFAULT_MAX_CONNECTIONS = 64;
|
||||
export const RUN_EGRESS_UPSTREAM_PROXY_ENV_KEY = 'VERITAS_EGRESS_UPSTREAM_PROXY' as const;
|
||||
export const RUN_EGRESS_PROXY_ENVIRONMENT_KEYS = [
|
||||
'HTTP_PROXY',
|
||||
'HTTPS_PROXY',
|
||||
|
|
@ -48,13 +49,36 @@ export interface RunEgressGatewayAuditEvent {
|
|||
decision: RunEgressDecision;
|
||||
}
|
||||
|
||||
export interface RunEgressGatewayApprovalRequest {
|
||||
gatewayId: string;
|
||||
runKey: string;
|
||||
protocol: RunEgressDecision['protocol'];
|
||||
host: string;
|
||||
port: number;
|
||||
method?: string;
|
||||
/** Path only. Query strings and fragments are never included. */
|
||||
path?: string;
|
||||
decision: RunEgressDecision;
|
||||
signal: AbortSignal;
|
||||
}
|
||||
|
||||
export interface RunEgressGatewayApprovalResult {
|
||||
approvalId: string;
|
||||
approved: boolean;
|
||||
}
|
||||
|
||||
export interface StartRunEgressGatewayInput {
|
||||
runId: string;
|
||||
policy: RunEgressPolicy;
|
||||
/** Optional operator-managed HTTP proxy. Credentials remain memory-only. */
|
||||
upstreamProxyUrl?: string;
|
||||
requestTimeoutMs?: number;
|
||||
idleTimeoutMs?: number;
|
||||
maxConnections?: number;
|
||||
onDecision?: (event: RunEgressGatewayAuditEvent) => Promise<void> | void;
|
||||
onApprovalRequired?: (
|
||||
request: RunEgressGatewayApprovalRequest
|
||||
) => Promise<RunEgressGatewayApprovalResult>;
|
||||
}
|
||||
|
||||
export interface RunEgressGatewayHandle {
|
||||
|
|
@ -67,9 +91,12 @@ export interface RunEgressGatewayHandle {
|
|||
interface ActiveGateway {
|
||||
runId: string;
|
||||
server: http.Server;
|
||||
socksServer: NetServer;
|
||||
sockets: Set<Socket>;
|
||||
evidence: RunEgressGatewayEvidence;
|
||||
environment: RunEgressGatewayHandle['environment'];
|
||||
abortController: AbortController;
|
||||
upstreamKey: string;
|
||||
stop(): Promise<RunEgressGatewayEvidence>;
|
||||
}
|
||||
|
||||
|
|
@ -85,10 +112,15 @@ export class RunEgressGatewayService {
|
|||
async start(input: StartRunEgressGatewayInput): Promise<RunEgressGatewayHandle> {
|
||||
const runId = input.runId.trim();
|
||||
if (!runId) throw new ConflictError('Run-scoped egress gateway requires a run id.');
|
||||
const upstream = parseUpstreamProxy(input.upstreamProxyUrl);
|
||||
const upstreamKey = upstream?.key ?? identity('upstream-proxy', 'direct');
|
||||
const existing = this.activeByRunId.get(runId);
|
||||
if (existing) {
|
||||
if (existing.evidence.policyHash !== input.policy.policyHash) {
|
||||
throw new ConflictError('Run egress gateway already uses a different policy.', {
|
||||
if (
|
||||
existing.evidence.policyHash !== input.policy.policyHash ||
|
||||
existing.upstreamKey !== upstreamKey
|
||||
) {
|
||||
throw new ConflictError('Run egress gateway already uses a different policy or upstream.', {
|
||||
gatewayId: existing.evidence.gatewayId,
|
||||
policyHash: existing.evidence.policyHash,
|
||||
});
|
||||
|
|
@ -117,7 +149,9 @@ export class RunEgressGatewayService {
|
|||
const maxConnections = boundedInteger(input.maxConnections, DEFAULT_MAX_CONNECTIONS, 1, 1_000);
|
||||
const sockets = new Set<Socket>();
|
||||
const clientSockets = new Set<Socket>();
|
||||
const abortController = new AbortController();
|
||||
const server = http.createServer();
|
||||
const socksServer = net.createServer();
|
||||
server.requestTimeout = requestTimeoutMs;
|
||||
server.headersTimeout = requestTimeoutMs;
|
||||
server.keepAliveTimeout = Math.min(idleTimeoutMs, requestTimeoutMs);
|
||||
|
|
@ -130,7 +164,10 @@ export class RunEgressGatewayService {
|
|||
requestTimeoutMs,
|
||||
idleTimeoutMs,
|
||||
policyService: this.policyService,
|
||||
upstream,
|
||||
onDecision: input.onDecision,
|
||||
onApprovalRequired: input.onApprovalRequired,
|
||||
signal: abortController.signal,
|
||||
now: this.now,
|
||||
sockets,
|
||||
};
|
||||
|
|
@ -144,45 +181,44 @@ export class RunEgressGatewayService {
|
|||
void handleUpgrade(context, request, socket as Socket, head);
|
||||
});
|
||||
server.on('connection', (socket) => {
|
||||
if (clientSockets.size >= maxConnections) {
|
||||
socket.destroy();
|
||||
registerClientSocket(socket, clientSockets, sockets, maxConnections, idleTimeoutMs);
|
||||
});
|
||||
socksServer.on('connection', (socket) => {
|
||||
if (!registerClientSocket(socket, clientSockets, sockets, maxConnections, idleTimeoutMs)) {
|
||||
return;
|
||||
}
|
||||
clientSockets.add(socket);
|
||||
sockets.add(socket);
|
||||
socket.setTimeout(idleTimeoutMs, () => socket.destroy());
|
||||
socket.once('close', () => {
|
||||
clientSockets.delete(socket);
|
||||
sockets.delete(socket);
|
||||
});
|
||||
void handleSocksConnection(context, socket);
|
||||
});
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onError = (error: Error) => {
|
||||
server.off('listening', onListening);
|
||||
reject(error);
|
||||
};
|
||||
const onListening = () => {
|
||||
server.off('error', onError);
|
||||
resolve();
|
||||
};
|
||||
server.once('error', onError);
|
||||
server.once('listening', onListening);
|
||||
server.listen(0, LOOPBACK_HOST);
|
||||
});
|
||||
await listenLoopback(server);
|
||||
const address = server.address();
|
||||
if (!address || typeof address === 'string') {
|
||||
server.close();
|
||||
throw new Error('Run egress gateway did not bind a TCP port.');
|
||||
}
|
||||
try {
|
||||
await listenLoopback(socksServer);
|
||||
} catch (error) {
|
||||
server.close();
|
||||
server.closeAllConnections?.();
|
||||
throw error;
|
||||
}
|
||||
const socksAddress = socksServer.address();
|
||||
if (!socksAddress || typeof socksAddress === 'string') {
|
||||
server.close();
|
||||
server.closeAllConnections?.();
|
||||
socksServer.close();
|
||||
throw new Error('Run egress SOCKS gateway did not bind a TCP port.');
|
||||
}
|
||||
const proxyUrl = `http://veritas:${encodeURIComponent(token)}@${LOOPBACK_HOST}:${address.port}`;
|
||||
const socksProxyUrl = `socks5h://veritas:${encodeURIComponent(token)}@${LOOPBACK_HOST}:${socksAddress.port}`;
|
||||
const environment: ActiveGateway['environment'] = {
|
||||
HTTP_PROXY: proxyUrl,
|
||||
HTTPS_PROXY: proxyUrl,
|
||||
ALL_PROXY: proxyUrl,
|
||||
ALL_PROXY: socksProxyUrl,
|
||||
http_proxy: proxyUrl,
|
||||
https_proxy: proxyUrl,
|
||||
all_proxy: proxyUrl,
|
||||
all_proxy: socksProxyUrl,
|
||||
NO_PROXY: '',
|
||||
no_proxy: '',
|
||||
};
|
||||
|
|
@ -193,7 +229,8 @@ export class RunEgressGatewayService {
|
|||
attributionKey: identity('attribution', token),
|
||||
policyHash: input.policy.policyHash,
|
||||
state: 'enforced',
|
||||
protocols: ['http', 'connect', 'ws'],
|
||||
protocols: ['http', 'connect', 'ws', 'socks5'],
|
||||
upstreamMode: upstream ? 'http-connect' : 'direct',
|
||||
proxyEnvironmentKeys: [...RUN_EGRESS_PROXY_ENVIRONMENT_KEYS],
|
||||
startedAt: this.now().toISOString(),
|
||||
};
|
||||
|
|
@ -201,9 +238,12 @@ export class RunEgressGatewayService {
|
|||
const active: ActiveGateway = {
|
||||
runId,
|
||||
server,
|
||||
socksServer,
|
||||
sockets,
|
||||
evidence,
|
||||
environment,
|
||||
abortController,
|
||||
upstreamKey,
|
||||
stop: () => {
|
||||
stopPromise ??= this.stopActive(active);
|
||||
return stopPromise;
|
||||
|
|
@ -235,11 +275,9 @@ export class RunEgressGatewayService {
|
|||
private async stopActive(active: ActiveGateway): Promise<RunEgressGatewayEvidence> {
|
||||
this.activeByRunId.delete(active.runId);
|
||||
this.activeByGatewayId.delete(active.evidence.gatewayId);
|
||||
active.abortController.abort(new Error('Run egress gateway stopped.'));
|
||||
for (const socket of active.sockets) socket.destroy();
|
||||
await new Promise<void>((resolve) => {
|
||||
active.server.close(() => resolve());
|
||||
active.server.closeAllConnections?.();
|
||||
});
|
||||
await Promise.all([closeServer(active.server), closeServer(active.socksServer)]);
|
||||
active.evidence = {
|
||||
...active.evidence,
|
||||
state: 'stopped',
|
||||
|
|
@ -267,11 +305,21 @@ interface GatewayRequestContext {
|
|||
requestTimeoutMs: number;
|
||||
idleTimeoutMs: number;
|
||||
policyService: EgressPolicyService;
|
||||
upstream?: UpstreamProxy;
|
||||
onDecision?: StartRunEgressGatewayInput['onDecision'];
|
||||
onApprovalRequired?: StartRunEgressGatewayInput['onApprovalRequired'];
|
||||
signal: AbortSignal;
|
||||
now: () => Date;
|
||||
sockets: Set<Socket>;
|
||||
}
|
||||
|
||||
interface UpstreamProxy {
|
||||
key: string;
|
||||
host: string;
|
||||
port: number;
|
||||
authorization?: string;
|
||||
}
|
||||
|
||||
async function handleHttpRequest(
|
||||
context: GatewayRequestContext,
|
||||
request: IncomingMessage,
|
||||
|
|
@ -293,9 +341,19 @@ async function handleHttpRequest(
|
|||
method: request.method,
|
||||
path: target.pathname,
|
||||
});
|
||||
await audit(context, resolution.decision);
|
||||
if (resolution.decision.decision === 'block') {
|
||||
rejectHttp(response, 403, resolution.decision.reason);
|
||||
const decision = await withApprovalIdlePause(request.socket, context.idleTimeoutMs, () =>
|
||||
authorizeBlockedRequest(context, {
|
||||
protocol: 'http',
|
||||
host: target.hostname,
|
||||
port: portFor(target, 80),
|
||||
method: request.method,
|
||||
path: target.pathname,
|
||||
decision: resolution.decision,
|
||||
})
|
||||
);
|
||||
await audit(context, decision);
|
||||
if (decision.decision === 'block') {
|
||||
rejectHttp(response, 403, decision.reason);
|
||||
return;
|
||||
}
|
||||
const address = resolution.resolvedAddresses[0];
|
||||
|
|
@ -303,6 +361,14 @@ async function handleHttpRequest(
|
|||
rejectHttp(response, 502, 'destination-resolution-failed');
|
||||
return;
|
||||
}
|
||||
let transport: Awaited<ReturnType<typeof openPinnedTransport>>;
|
||||
try {
|
||||
transport = await openPinnedTransport(context, address, portFor(target, 80));
|
||||
} catch {
|
||||
rejectHttp(response, 502, 'upstream-failure');
|
||||
return;
|
||||
}
|
||||
if (transport.head.length > 0) transport.socket.unshift(transport.head);
|
||||
const headers = forwardHeaders(request.headers, target.host, false);
|
||||
const upstream = http.request({
|
||||
host: address,
|
||||
|
|
@ -312,10 +378,8 @@ async function handleHttpRequest(
|
|||
headers,
|
||||
family: net.isIP(address),
|
||||
timeout: context.requestTimeoutMs,
|
||||
});
|
||||
upstream.once('socket', (socket) => {
|
||||
context.sockets.add(socket);
|
||||
socket.once('close', () => context.sockets.delete(socket));
|
||||
agent: false,
|
||||
createConnection: () => transport.socket,
|
||||
});
|
||||
upstream.once('response', (upstreamResponse) => {
|
||||
response.writeHead(
|
||||
|
|
@ -352,9 +416,17 @@ async function handleConnect(
|
|||
host: authority.host,
|
||||
port: authority.port,
|
||||
});
|
||||
await audit(context, resolution.decision);
|
||||
if (resolution.decision.decision === 'block') {
|
||||
rejectSocket(client, 403, resolution.decision.reason);
|
||||
const decision = await withApprovalIdlePause(client, context.idleTimeoutMs, () =>
|
||||
authorizeBlockedRequest(context, {
|
||||
protocol: 'https',
|
||||
host: authority.host,
|
||||
port: authority.port,
|
||||
decision: resolution.decision,
|
||||
})
|
||||
);
|
||||
await audit(context, decision);
|
||||
if (decision.decision === 'block') {
|
||||
rejectSocket(client, 403, decision.reason);
|
||||
return;
|
||||
}
|
||||
const address = resolution.resolvedAddresses[0];
|
||||
|
|
@ -362,26 +434,20 @@ async function handleConnect(
|
|||
rejectSocket(client, 502, 'destination-resolution-failed');
|
||||
return;
|
||||
}
|
||||
const upstream = net.connect({
|
||||
host: address,
|
||||
port: authority.port,
|
||||
family: net.isIP(address),
|
||||
});
|
||||
context.sockets.add(upstream);
|
||||
let connected = false;
|
||||
upstream.setTimeout(context.idleTimeoutMs, () => upstream.destroy());
|
||||
upstream.once('close', () => context.sockets.delete(upstream));
|
||||
upstream.once('connect', () => {
|
||||
connected = true;
|
||||
let transport: Awaited<ReturnType<typeof openPinnedTransport>>;
|
||||
try {
|
||||
transport = await openPinnedTransport(context, address, authority.port);
|
||||
} catch {
|
||||
rejectSocket(client, 502, 'upstream-failure');
|
||||
return;
|
||||
}
|
||||
const upstream = transport.socket;
|
||||
client.write('HTTP/1.1 200 Connection Established\r\n\r\n');
|
||||
if (transport.head.length > 0) client.write(transport.head);
|
||||
if (head.length > 0) upstream.write(head);
|
||||
client.pipe(upstream);
|
||||
upstream.pipe(client);
|
||||
});
|
||||
upstream.once('error', () => {
|
||||
if (connected) client.destroy();
|
||||
else rejectSocket(client, 502, 'upstream-failure');
|
||||
});
|
||||
upstream.once('error', () => client.destroy());
|
||||
}
|
||||
|
||||
async function handleUpgrade(
|
||||
|
|
@ -411,9 +477,19 @@ async function handleUpgrade(
|
|||
method: request.method ?? 'GET',
|
||||
path: target.pathname,
|
||||
});
|
||||
await audit(context, resolution.decision);
|
||||
if (resolution.decision.decision === 'block') {
|
||||
rejectSocket(client, 403, resolution.decision.reason);
|
||||
const decision = await withApprovalIdlePause(client, context.idleTimeoutMs, () =>
|
||||
authorizeBlockedRequest(context, {
|
||||
protocol: 'ws',
|
||||
host: target.hostname,
|
||||
port: portFor(target, 80),
|
||||
method: request.method ?? 'GET',
|
||||
path: target.pathname,
|
||||
decision: resolution.decision,
|
||||
})
|
||||
);
|
||||
await audit(context, decision);
|
||||
if (decision.decision === 'block') {
|
||||
rejectSocket(client, 403, decision.reason);
|
||||
return;
|
||||
}
|
||||
const address = resolution.resolvedAddresses[0];
|
||||
|
|
@ -421,36 +497,223 @@ async function handleUpgrade(
|
|||
rejectSocket(client, 502, 'destination-resolution-failed');
|
||||
return;
|
||||
}
|
||||
const upstream = net.connect({
|
||||
host: address,
|
||||
port: portFor(target, 80),
|
||||
family: net.isIP(address),
|
||||
});
|
||||
context.sockets.add(upstream);
|
||||
let connected = false;
|
||||
upstream.setTimeout(context.idleTimeoutMs, () => upstream.destroy());
|
||||
upstream.once('close', () => context.sockets.delete(upstream));
|
||||
upstream.once('connect', () => {
|
||||
connected = true;
|
||||
let transport: Awaited<ReturnType<typeof openPinnedTransport>>;
|
||||
try {
|
||||
transport = await openPinnedTransport(context, address, portFor(target, 80));
|
||||
} catch {
|
||||
rejectSocket(client, 502, 'upstream-failure');
|
||||
return;
|
||||
}
|
||||
const upstream = transport.socket;
|
||||
upstream.write(
|
||||
`${request.method ?? 'GET'} ${target.pathname}${target.search} HTTP/${request.httpVersion}\r\n`
|
||||
);
|
||||
for (const [name, value] of Object.entries(
|
||||
forwardHeaders(request.headers, target.host, true)
|
||||
)) {
|
||||
for (const [name, value] of Object.entries(forwardHeaders(request.headers, target.host, true))) {
|
||||
for (const item of Array.isArray(value) ? value : [value]) {
|
||||
if (item !== undefined) upstream.write(`${name}: ${item}\r\n`);
|
||||
}
|
||||
}
|
||||
upstream.write('\r\n');
|
||||
if (transport.head.length > 0) client.write(transport.head);
|
||||
if (head.length > 0) upstream.write(head);
|
||||
client.pipe(upstream);
|
||||
upstream.pipe(client);
|
||||
upstream.once('error', () => client.destroy());
|
||||
}
|
||||
|
||||
async function handleSocksConnection(
|
||||
context: GatewayRequestContext,
|
||||
client: Socket
|
||||
): Promise<void> {
|
||||
const reader = new SocketByteReader(client);
|
||||
try {
|
||||
const greeting = await reader.readExactly(2);
|
||||
if (greeting[0] !== 0x05 || greeting[1] === 0) {
|
||||
client.end(Buffer.from([0x05, 0xff]));
|
||||
return;
|
||||
}
|
||||
const methods = await reader.readExactly(greeting[1]);
|
||||
if (!methods.includes(0x02)) {
|
||||
client.end(Buffer.from([0x05, 0xff]));
|
||||
return;
|
||||
}
|
||||
client.write(Buffer.from([0x05, 0x02]));
|
||||
|
||||
const authHeader = await reader.readExactly(2);
|
||||
if (authHeader[0] !== 0x01 || authHeader[1] === 0) {
|
||||
client.end(Buffer.from([0x01, 0x01]));
|
||||
return;
|
||||
}
|
||||
const username = (await reader.readExactly(authHeader[1])).toString('utf8');
|
||||
const passwordLength = (await reader.readExactly(1))[0] ?? 0;
|
||||
if (passwordLength === 0) {
|
||||
client.end(Buffer.from([0x01, 0x01]));
|
||||
return;
|
||||
}
|
||||
const password = (await reader.readExactly(passwordLength)).toString('utf8');
|
||||
if (!safeEqual(username, 'veritas') || !safeEqual(password, context.token)) {
|
||||
client.end(Buffer.from([0x01, 0x01]));
|
||||
return;
|
||||
}
|
||||
client.write(Buffer.from([0x01, 0x00]));
|
||||
|
||||
const requestHeader = await reader.readExactly(4);
|
||||
if (requestHeader[0] !== 0x05 || requestHeader[2] !== 0x00) {
|
||||
endSocks(client, 0x01);
|
||||
return;
|
||||
}
|
||||
if (requestHeader[1] !== 0x01) {
|
||||
endSocks(client, 0x07);
|
||||
return;
|
||||
}
|
||||
const host = await readSocksHost(reader, requestHeader[3] ?? 0);
|
||||
if (!host) {
|
||||
endSocks(client, 0x08);
|
||||
return;
|
||||
}
|
||||
const port = (await reader.readExactly(2)).readUInt16BE(0);
|
||||
const resolution = await context.policyService.resolveAndEvaluate(context.policy, {
|
||||
protocol: 'socks',
|
||||
host,
|
||||
port,
|
||||
});
|
||||
upstream.once('error', () => {
|
||||
if (connected) client.destroy();
|
||||
else rejectSocket(client, 502, 'upstream-failure');
|
||||
const decision = await withApprovalIdlePause(client, context.idleTimeoutMs, () =>
|
||||
authorizeBlockedRequest(context, {
|
||||
protocol: 'socks',
|
||||
host,
|
||||
port,
|
||||
decision: resolution.decision,
|
||||
})
|
||||
);
|
||||
await audit(context, decision);
|
||||
if (decision.decision === 'block') {
|
||||
endSocks(client, 0x02);
|
||||
return;
|
||||
}
|
||||
const address = resolution.resolvedAddresses[0];
|
||||
if (!address) {
|
||||
endSocks(client, 0x04);
|
||||
return;
|
||||
}
|
||||
|
||||
const transport = await openPinnedTransport(context, address, port);
|
||||
const upstream = transport.socket;
|
||||
client.write(socksReply(0x00));
|
||||
const remainder = reader.release();
|
||||
client.on('error', () => client.destroy());
|
||||
upstream.on('error', () => client.destroy());
|
||||
if (transport.head.length > 0) client.write(transport.head);
|
||||
if (remainder.length > 0) upstream.write(remainder);
|
||||
client.pipe(upstream);
|
||||
upstream.pipe(client);
|
||||
} catch {
|
||||
reader.release();
|
||||
if (!client.destroyed) endSocks(client, 0x01);
|
||||
}
|
||||
}
|
||||
|
||||
class SocketByteReader {
|
||||
private buffer = Buffer.alloc(0);
|
||||
private requestedBytes = 0;
|
||||
private resolveRead?: (value: Buffer) => void;
|
||||
private rejectRead?: (error: Error) => void;
|
||||
private released = false;
|
||||
|
||||
constructor(private readonly socket: Socket) {
|
||||
socket.on('data', this.onData);
|
||||
socket.once('close', this.onClose);
|
||||
}
|
||||
|
||||
readExactly(size: number): Promise<Buffer> {
|
||||
if (!Number.isSafeInteger(size) || size < 1 || size > 65_535) {
|
||||
return Promise.reject(new Error('Invalid SOCKS frame length.'));
|
||||
}
|
||||
if (this.released || this.resolveRead) {
|
||||
return Promise.reject(new Error('SOCKS frame reader is unavailable.'));
|
||||
}
|
||||
if (this.buffer.length >= size) return Promise.resolve(this.take(size));
|
||||
this.requestedBytes = size;
|
||||
return new Promise<Buffer>((resolve, reject) => {
|
||||
this.resolveRead = resolve;
|
||||
this.rejectRead = reject;
|
||||
});
|
||||
}
|
||||
|
||||
release(): Buffer {
|
||||
if (this.released) return Buffer.alloc(0);
|
||||
this.released = true;
|
||||
this.socket.off('data', this.onData);
|
||||
this.socket.off('close', this.onClose);
|
||||
const buffered = this.buffer;
|
||||
this.buffer = Buffer.alloc(0);
|
||||
return buffered;
|
||||
}
|
||||
|
||||
private readonly onData = (chunk: Buffer): void => {
|
||||
if (this.released) return;
|
||||
this.buffer = Buffer.concat([this.buffer, chunk]);
|
||||
if (this.buffer.length > 65_535) {
|
||||
this.rejectPending(new Error('SOCKS handshake exceeded the supported size.'));
|
||||
this.socket.destroy();
|
||||
return;
|
||||
}
|
||||
if (!this.resolveRead || this.buffer.length < this.requestedBytes) return;
|
||||
const value = this.take(this.requestedBytes);
|
||||
const resolve = this.resolveRead;
|
||||
this.resolveRead = undefined;
|
||||
this.rejectRead = undefined;
|
||||
this.requestedBytes = 0;
|
||||
resolve(value);
|
||||
};
|
||||
|
||||
private readonly onClose = (): void => {
|
||||
this.rejectPending(new Error('SOCKS client disconnected during negotiation.'));
|
||||
};
|
||||
|
||||
private take(size: number): Buffer {
|
||||
const value = this.buffer.subarray(0, size);
|
||||
this.buffer = this.buffer.subarray(size);
|
||||
return value;
|
||||
}
|
||||
|
||||
private rejectPending(error: Error): void {
|
||||
const reject = this.rejectRead;
|
||||
this.resolveRead = undefined;
|
||||
this.rejectRead = undefined;
|
||||
this.requestedBytes = 0;
|
||||
reject?.(error);
|
||||
}
|
||||
}
|
||||
|
||||
async function readSocksHost(
|
||||
reader: SocketByteReader,
|
||||
addressType: number
|
||||
): Promise<string | null> {
|
||||
if (addressType === 0x01) {
|
||||
return [...(await reader.readExactly(4))].join('.');
|
||||
}
|
||||
if (addressType === 0x03) {
|
||||
const length = (await reader.readExactly(1))[0] ?? 0;
|
||||
if (length === 0) return null;
|
||||
return (await reader.readExactly(length)).toString('utf8').toLowerCase();
|
||||
}
|
||||
if (addressType === 0x04) {
|
||||
const address = await reader.readExactly(16);
|
||||
const groups: string[] = [];
|
||||
for (let offset = 0; offset < address.length; offset += 2) {
|
||||
groups.push(address.readUInt16BE(offset).toString(16));
|
||||
}
|
||||
return groups.join(':');
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function socksReply(code: number): Buffer {
|
||||
return Buffer.from([0x05, code, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]);
|
||||
}
|
||||
|
||||
function endSocks(socket: Socket, code: number): void {
|
||||
if (!socket.destroyed) socket.end(socksReply(code));
|
||||
}
|
||||
|
||||
async function audit(context: GatewayRequestContext, decision: RunEgressDecision): Promise<void> {
|
||||
|
|
@ -467,6 +730,144 @@ async function audit(context: GatewayRequestContext, decision: RunEgressDecision
|
|||
}
|
||||
}
|
||||
|
||||
async function authorizeBlockedRequest(
|
||||
context: GatewayRequestContext,
|
||||
input: Omit<RunEgressGatewayApprovalRequest, 'gatewayId' | 'runKey' | 'signal'>
|
||||
): Promise<RunEgressDecision> {
|
||||
if (
|
||||
input.decision.decision !== 'block' ||
|
||||
!input.decision.approvalEligible ||
|
||||
!context.onApprovalRequired ||
|
||||
context.signal.aborted
|
||||
) {
|
||||
return input.decision;
|
||||
}
|
||||
const policyReason = input.decision.reason;
|
||||
if (
|
||||
policyReason === 'allowed-by-default' ||
|
||||
policyReason === 'allowed-by-host-rule' ||
|
||||
policyReason === 'allowed-by-approval'
|
||||
) {
|
||||
return input.decision;
|
||||
}
|
||||
try {
|
||||
const result = await context.onApprovalRequired({
|
||||
gatewayId: context.gatewayId,
|
||||
runKey: context.runKey,
|
||||
...input,
|
||||
signal: context.signal,
|
||||
});
|
||||
if (!result.approved) {
|
||||
return { ...input.decision, approvalId: result.approvalId };
|
||||
}
|
||||
return {
|
||||
...input.decision,
|
||||
decision: 'allow',
|
||||
reason: 'allowed-by-approval',
|
||||
approvalEligible: false,
|
||||
approvalId: result.approvalId,
|
||||
policyReason,
|
||||
};
|
||||
} catch {
|
||||
return input.decision;
|
||||
}
|
||||
}
|
||||
|
||||
async function withApprovalIdlePause<T>(
|
||||
socket: Socket,
|
||||
idleTimeoutMs: number,
|
||||
action: () => Promise<T>
|
||||
): Promise<T> {
|
||||
socket.setTimeout(0);
|
||||
try {
|
||||
return await action();
|
||||
} finally {
|
||||
if (!socket.destroyed) socket.setTimeout(idleTimeoutMs, () => socket.destroy());
|
||||
}
|
||||
}
|
||||
|
||||
async function openPinnedTransport(
|
||||
context: GatewayRequestContext,
|
||||
address: string,
|
||||
port: number
|
||||
): Promise<{ socket: Socket; head: Buffer }> {
|
||||
const upstream = context.upstream;
|
||||
const socket = net.connect({
|
||||
host: upstream?.host ?? address,
|
||||
port: upstream?.port ?? port,
|
||||
...(upstream ? {} : { family: net.isIP(address) }),
|
||||
});
|
||||
context.sockets.add(socket);
|
||||
socket.setTimeout(context.idleTimeoutMs, () => socket.destroy());
|
||||
socket.once('close', () => context.sockets.delete(socket));
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
socket.once('connect', resolve);
|
||||
socket.once('error', reject);
|
||||
});
|
||||
if (!upstream) return { socket, head: Buffer.alloc(0) };
|
||||
|
||||
const authority = formatAuthority(address, port);
|
||||
socket.write(
|
||||
`CONNECT ${authority} HTTP/1.1\r\n` +
|
||||
`Host: ${authority}\r\n` +
|
||||
'Connection: keep-alive\r\n' +
|
||||
(upstream.authorization ? `Proxy-Authorization: ${upstream.authorization}\r\n` : '') +
|
||||
'\r\n'
|
||||
);
|
||||
const response = await readHttpResponseHead(socket);
|
||||
if (response.statusCode !== 200) {
|
||||
socket.destroy();
|
||||
throw new Error(`Upstream proxy rejected CONNECT with status ${response.statusCode}.`);
|
||||
}
|
||||
return { socket, head: response.head };
|
||||
}
|
||||
|
||||
function readHttpResponseHead(socket: Socket): Promise<{ statusCode: number; head: Buffer }> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let buffer = Buffer.alloc(0);
|
||||
const cleanup = () => {
|
||||
socket.off('data', onData);
|
||||
socket.off('close', onClose);
|
||||
socket.off('error', onError);
|
||||
};
|
||||
const onClose = () => {
|
||||
cleanup();
|
||||
reject(new Error('Upstream proxy disconnected during CONNECT.'));
|
||||
};
|
||||
const onError = (error: Error) => {
|
||||
cleanup();
|
||||
reject(error);
|
||||
};
|
||||
const onData = (chunk: Buffer) => {
|
||||
buffer = Buffer.concat([buffer, chunk]);
|
||||
if (buffer.length > 16 * 1024) {
|
||||
cleanup();
|
||||
socket.destroy();
|
||||
reject(new Error('Upstream proxy response headers exceeded the supported size.'));
|
||||
return;
|
||||
}
|
||||
const boundary = buffer.indexOf('\r\n\r\n');
|
||||
if (boundary < 0) return;
|
||||
const firstLine = buffer.subarray(0, boundary).toString('latin1').split('\r\n', 1)[0] ?? '';
|
||||
const match = /^HTTP\/1\.[01] ([1-5]\d{2})(?:\s|$)/.exec(firstLine);
|
||||
socket.pause();
|
||||
cleanup();
|
||||
if (!match) {
|
||||
socket.destroy();
|
||||
reject(new Error('Upstream proxy returned an invalid CONNECT response.'));
|
||||
return;
|
||||
}
|
||||
resolve({
|
||||
statusCode: Number(match[1]),
|
||||
head: buffer.subarray(boundary + 4),
|
||||
});
|
||||
};
|
||||
socket.on('data', onData);
|
||||
socket.once('close', onClose);
|
||||
socket.once('error', onError);
|
||||
});
|
||||
}
|
||||
|
||||
function authenticated(request: IncomingMessage, token: string): boolean {
|
||||
const header = request.headers['proxy-authorization'];
|
||||
if (typeof header !== 'string') return false;
|
||||
|
|
@ -481,6 +882,54 @@ function safeEqual(left: string, right: string): boolean {
|
|||
return leftDigest.equals(rightDigest);
|
||||
}
|
||||
|
||||
function parseUpstreamProxy(value: string | undefined): UpstreamProxy | undefined {
|
||||
const source = value?.trim();
|
||||
if (!source) return undefined;
|
||||
let parsed: URL;
|
||||
try {
|
||||
parsed = new URL(source);
|
||||
} catch {
|
||||
throw new ConflictError('Run egress upstream proxy URL is invalid.');
|
||||
}
|
||||
if (
|
||||
parsed.protocol !== 'http:' ||
|
||||
parsed.pathname !== '/' ||
|
||||
parsed.search ||
|
||||
parsed.hash ||
|
||||
!parsed.hostname
|
||||
) {
|
||||
throw new ConflictError(
|
||||
'Run egress upstream proxy must be an HTTP origin without a path, query, or fragment.'
|
||||
);
|
||||
}
|
||||
const port = parsed.port ? Number.parseInt(parsed.port, 10) : 80;
|
||||
if (!Number.isInteger(port) || port < 1 || port > 65_535) {
|
||||
throw new ConflictError('Run egress upstream proxy port is invalid.');
|
||||
}
|
||||
let username: string;
|
||||
let password: string;
|
||||
try {
|
||||
username = decodeURIComponent(parsed.username);
|
||||
password = decodeURIComponent(parsed.password);
|
||||
} catch {
|
||||
throw new ConflictError('Run egress upstream proxy credentials have invalid encoding.');
|
||||
}
|
||||
return {
|
||||
key: identity('upstream-proxy', parsed.toString()),
|
||||
host: parsed.hostname.replace(/^\[|\]$/g, ''),
|
||||
port,
|
||||
...(username || password
|
||||
? {
|
||||
authorization: `Basic ${Buffer.from(`${username}:${password}`).toString('base64')}`,
|
||||
}
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
function formatAuthority(host: string, port: number): string {
|
||||
return `${net.isIP(host) === 6 ? `[${host}]` : host}:${port}`;
|
||||
}
|
||||
|
||||
function parseProxyUrl(value: string | undefined): URL | null {
|
||||
if (!value) return null;
|
||||
try {
|
||||
|
|
@ -547,6 +996,50 @@ function rejectSocket(socket: Socket, statusCode: number, reason: string): void
|
|||
);
|
||||
}
|
||||
|
||||
function registerClientSocket(
|
||||
socket: Socket,
|
||||
clientSockets: Set<Socket>,
|
||||
sockets: Set<Socket>,
|
||||
maxConnections: number,
|
||||
idleTimeoutMs: number
|
||||
): boolean {
|
||||
if (clientSockets.size >= maxConnections) {
|
||||
socket.destroy();
|
||||
return false;
|
||||
}
|
||||
clientSockets.add(socket);
|
||||
sockets.add(socket);
|
||||
socket.setTimeout(idleTimeoutMs, () => socket.destroy());
|
||||
socket.once('close', () => {
|
||||
clientSockets.delete(socket);
|
||||
sockets.delete(socket);
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
function listenLoopback(server: http.Server | NetServer): Promise<void> {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const onError = (error: Error) => {
|
||||
server.off('listening', onListening);
|
||||
reject(error);
|
||||
};
|
||||
const onListening = () => {
|
||||
server.off('error', onError);
|
||||
resolve();
|
||||
};
|
||||
server.once('error', onError);
|
||||
server.once('listening', onListening);
|
||||
server.listen(0, LOOPBACK_HOST);
|
||||
});
|
||||
}
|
||||
|
||||
function closeServer(server: http.Server | NetServer): Promise<void> {
|
||||
return new Promise<void>((resolve) => {
|
||||
server.close(() => resolve());
|
||||
if ('closeAllConnections' in server) server.closeAllConnections();
|
||||
});
|
||||
}
|
||||
|
||||
function boundedDuration(
|
||||
value: number | undefined,
|
||||
fallback: number,
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import type {
|
|||
AgentConfig,
|
||||
AgentHostRoutingDecision,
|
||||
ExecutableAgentProvider,
|
||||
NetworkEgressTelemetryEvent,
|
||||
ProviderRuntimeCapabilityId,
|
||||
ProviderRuntimeManifest,
|
||||
RunLaunchPhaseAuthority,
|
||||
|
|
@ -43,9 +44,17 @@ import { getAgentHostService } from './agent-host-service.js';
|
|||
import { getSandboxPolicyService } from './sandbox-policy-service.js';
|
||||
import {
|
||||
getRunEgressGatewayService,
|
||||
RUN_EGRESS_UPSTREAM_PROXY_ENV_KEY,
|
||||
runEgressPolicyRequiresGateway,
|
||||
type RunEgressGatewayApprovalRequest,
|
||||
type RunEgressGatewayApprovalResult,
|
||||
type RunEgressGatewayService,
|
||||
} from './run-egress-gateway-service.js';
|
||||
import {
|
||||
getRunApprovalBrokerService,
|
||||
type RunApprovalBrokerService,
|
||||
} from './run-approval-broker-service.js';
|
||||
import { getTelemetryService } from './telemetry-service.js';
|
||||
import {
|
||||
assertProviderRuntimeCapabilities,
|
||||
assertProviderRuntimeControl,
|
||||
|
|
@ -86,6 +95,7 @@ interface WorkflowStepExecutorOptions {
|
|||
persistRun?: (run: WorkflowRun) => Promise<void>;
|
||||
phaseAuthority?: PhaseLaunchAuthorityService;
|
||||
runEgressGateway?: Pick<RunEgressGatewayService, 'start'>;
|
||||
approvalBroker?: Pick<RunApprovalBrokerService, 'request' | 'awaitDecision' | 'cancelAttempt'>;
|
||||
}
|
||||
|
||||
type WorkflowExecutableProvider = Extract<ExecutableAgentProvider, 'codex-sdk' | 'openclaw'>;
|
||||
|
|
@ -129,6 +139,10 @@ export class WorkflowStepExecutor {
|
|||
private persistRun: (run: WorkflowRun) => Promise<void>;
|
||||
private phaseAuthority: PhaseLaunchAuthorityService;
|
||||
private runEgressGateway: Pick<RunEgressGatewayService, 'start'>;
|
||||
private approvalBroker: Pick<
|
||||
RunApprovalBrokerService,
|
||||
'request' | 'awaitDecision' | 'cancelAttempt'
|
||||
>;
|
||||
private appendCountCache?: Map<string, number>; // Performance: Track append counts to reduce stat() calls
|
||||
|
||||
constructor(runsDir?: string, options: WorkflowStepExecutorOptions = {}) {
|
||||
|
|
@ -143,6 +157,7 @@ export class WorkflowStepExecutor {
|
|||
this.persistRun = options.persistRun ?? (async () => undefined);
|
||||
this.phaseAuthority = options.phaseAuthority ?? new PhaseLaunchAuthorityService();
|
||||
this.runEgressGateway = options.runEgressGateway ?? getRunEgressGatewayService();
|
||||
this.approvalBroker = options.approvalBroker ?? getRunApprovalBrokerService();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -626,6 +641,96 @@ export class WorkflowStepExecutor {
|
|||
return `${run.id}:${stepId}:${sequence}`;
|
||||
}
|
||||
|
||||
private workflowTaskId(run: WorkflowRun): string {
|
||||
const contextTask = run.context.task as { id?: string } | undefined;
|
||||
return run.taskId ?? contextTask?.id ?? run.id;
|
||||
}
|
||||
|
||||
private workflowProject(run: WorkflowRun): string | undefined {
|
||||
const contextTask = run.context.task as { project?: string } | undefined;
|
||||
return contextTask?.project;
|
||||
}
|
||||
|
||||
private async resolveWorkflowEgressApproval(
|
||||
run: WorkflowRun,
|
||||
step: WorkflowStep,
|
||||
agentId: string,
|
||||
runtimeManifest: ProviderRuntimeManifest,
|
||||
sandboxPolicy: SandboxPolicyDryRunResult,
|
||||
request: RunEgressGatewayApprovalRequest
|
||||
): Promise<RunEgressGatewayApprovalResult> {
|
||||
const taskId = this.workflowTaskId(run);
|
||||
const attemptId = this.workflowStepAttemptId(run, step.id);
|
||||
const providerRequestId = `egress:${digestRunLaunchValue({
|
||||
gatewayId: request.gatewayId,
|
||||
protocol: request.protocol,
|
||||
host: request.host,
|
||||
port: request.port,
|
||||
method: request.method,
|
||||
path: request.path,
|
||||
policyHash: request.decision.policyHash,
|
||||
}).slice('sha256:'.length, 'sha256:'.length + 32)}`;
|
||||
const approval = await this.approvalBroker.request({
|
||||
workspaceId: 'local',
|
||||
taskId,
|
||||
attemptId,
|
||||
provider: 'codex-sdk',
|
||||
agentId,
|
||||
providerRequestId,
|
||||
requestKind: 'approval',
|
||||
actionClass: 'network',
|
||||
action: `Allow ${request.protocol} egress to ${request.host}:${request.port}`,
|
||||
details: `Blocked by ${request.decision.reason}; method ${request.method ?? 'not available'}.`,
|
||||
resourceScope: [request.host, `${request.protocol}:${request.port}`],
|
||||
workingDirectory: this.expandPath(this.getWorkflowWorkingDirectory(run)),
|
||||
riskClass: 'high',
|
||||
policyReason: request.decision.reason,
|
||||
evidenceRevision: digestRunLaunchValue({
|
||||
runtimeManifestDigest: runtimeManifest.digest,
|
||||
policyHash: sandboxPolicy.effective.networkPolicy?.policyHash,
|
||||
}),
|
||||
mobileSafe: false,
|
||||
exactAction: {
|
||||
protocol: request.protocol,
|
||||
host: request.host,
|
||||
port: request.port,
|
||||
method: request.method,
|
||||
path: request.path,
|
||||
policyHash: request.decision.policyHash,
|
||||
blockedReason: request.decision.reason,
|
||||
},
|
||||
});
|
||||
if (request.signal.aborted) {
|
||||
await this.approvalBroker.cancelAttempt(
|
||||
'local',
|
||||
taskId,
|
||||
attemptId,
|
||||
'Workflow egress gateway stopped.'
|
||||
);
|
||||
return { approvalId: approval.id, approved: false };
|
||||
}
|
||||
try {
|
||||
const resolved = await this.approvalBroker.awaitDecision(approval.id, {
|
||||
signal: request.signal,
|
||||
});
|
||||
return {
|
||||
approvalId: approval.id,
|
||||
approved: resolved.request.status === 'approved',
|
||||
};
|
||||
} catch (error) {
|
||||
if (request.signal.aborted) {
|
||||
await this.approvalBroker.cancelAttempt(
|
||||
'local',
|
||||
taskId,
|
||||
attemptId,
|
||||
'Workflow egress gateway stopped.'
|
||||
);
|
||||
return { approvalId: approval.id, approved: false };
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private recordAgentHostRouting(
|
||||
run: WorkflowRun,
|
||||
step: WorkflowStep,
|
||||
|
|
@ -863,11 +968,24 @@ export class WorkflowStepExecutor {
|
|||
`Workflow agent ${agentDef?.id || step.agent || step.id}`
|
||||
);
|
||||
const egressPolicy = sandboxPolicy.effective.networkPolicy;
|
||||
const attemptId = this.workflowStepAttemptId(run, step.id);
|
||||
const taskId = this.workflowTaskId(run);
|
||||
const agentId = step.agent ?? agentDef?.id ?? step.id;
|
||||
const egressGateway =
|
||||
egressPolicy && runEgressPolicyRequiresGateway(egressPolicy)
|
||||
? await this.runEgressGateway.start({
|
||||
runId: this.workflowStepAttemptId(run, step.id),
|
||||
runId: attemptId,
|
||||
policy: egressPolicy,
|
||||
upstreamProxyUrl: process.env[RUN_EGRESS_UPSTREAM_PROXY_ENV_KEY],
|
||||
onApprovalRequired: (request) =>
|
||||
this.resolveWorkflowEgressApproval(
|
||||
run,
|
||||
step,
|
||||
agentId,
|
||||
runtimeManifest,
|
||||
sandboxPolicy,
|
||||
request
|
||||
),
|
||||
onDecision: async (event) => {
|
||||
await getGovernanceTraceService().record({
|
||||
kind: 'policy',
|
||||
|
|
@ -875,7 +993,7 @@ export class WorkflowStepExecutor {
|
|||
title: 'Workflow egress decision',
|
||||
summary: `${event.decision.protocol} request ${event.decision.decision}: ${event.decision.reason}`,
|
||||
subject: {
|
||||
taskId: run.taskId,
|
||||
taskId,
|
||||
agentId: step.agent,
|
||||
actionType: 'workflow.network-egress',
|
||||
},
|
||||
|
|
@ -902,8 +1020,31 @@ export class WorkflowStepExecutor {
|
|||
reason: event.decision.reason,
|
||||
blockedAddressClass: event.decision.blockedAddressClass,
|
||||
approvalEligible: event.decision.approvalEligible,
|
||||
approvalId: event.decision.approvalId,
|
||||
policyReason: event.decision.policyReason,
|
||||
},
|
||||
});
|
||||
await getTelemetryService().emit<NetworkEgressTelemetryEvent>({
|
||||
type: 'network.egress',
|
||||
taskId,
|
||||
attemptId,
|
||||
agent: agentId,
|
||||
provider: 'codex-sdk',
|
||||
project: this.workflowProject(run),
|
||||
gatewayId: event.gatewayId,
|
||||
runKey: event.runKey,
|
||||
policyHash: event.decision.policyHash,
|
||||
protocol: event.decision.protocol,
|
||||
hostKey: event.decision.hostKey,
|
||||
port: event.decision.port,
|
||||
method: event.decision.method,
|
||||
decision: event.decision.decision,
|
||||
reason: event.decision.reason,
|
||||
policyReason: event.decision.policyReason,
|
||||
blockedAddressClass: event.decision.blockedAddressClass,
|
||||
approvalEligible: event.decision.approvalEligible,
|
||||
approvalId: event.decision.approvalId,
|
||||
});
|
||||
},
|
||||
})
|
||||
: undefined;
|
||||
|
|
@ -912,10 +1053,12 @@ export class WorkflowStepExecutor {
|
|||
const workingDirectory = this.expandPath(this.getWorkflowWorkingDirectory(run));
|
||||
const codex = new Codex({
|
||||
codexPathOverride,
|
||||
env: {
|
||||
env: Object.fromEntries(
|
||||
Object.entries({
|
||||
...buildSafeCodexEnv(process.env, sandboxPolicy.effective.envPassthrough),
|
||||
...egressGateway?.environment,
|
||||
},
|
||||
}).filter(([key]) => key !== RUN_EGRESS_UPSTREAM_PROXY_ENV_KEY)
|
||||
),
|
||||
});
|
||||
|
||||
const sessionKey = step.agent || agentDef?.id || step.id;
|
||||
|
|
|
|||
|
|
@ -87,6 +87,7 @@ export interface RunEgressDecisionInput {
|
|||
export type RunEgressDecisionReason =
|
||||
| 'allowed-by-default'
|
||||
| 'allowed-by-host-rule'
|
||||
| 'allowed-by-approval'
|
||||
| 'default-deny'
|
||||
| 'denied-host'
|
||||
| 'method-not-allowed'
|
||||
|
|
@ -109,6 +110,13 @@ export interface RunEgressDecision {
|
|||
matchedRule?: RunEgressHostRule;
|
||||
blockedAddressClass?: 'private' | 'loopback' | 'link-local' | 'metadata';
|
||||
approvalEligible: boolean;
|
||||
/** Durable approval correlated to this decision without exposing the destination URL. */
|
||||
approvalId?: string;
|
||||
/** Original blocking reason when a scoped approval changed the transport decision. */
|
||||
policyReason?: Exclude<
|
||||
RunEgressDecisionReason,
|
||||
'allowed-by-default' | 'allowed-by-host-rule' | 'allowed-by-approval'
|
||||
>;
|
||||
}
|
||||
|
||||
export interface RunEgressGatewayEvidence {
|
||||
|
|
@ -118,7 +126,9 @@ export interface RunEgressGatewayEvidence {
|
|||
attributionKey: string;
|
||||
policyHash: string;
|
||||
state: 'enforced' | 'stopped';
|
||||
protocols: Array<'http' | 'connect' | 'ws'>;
|
||||
protocols: Array<'http' | 'connect' | 'ws' | 'socks5'>;
|
||||
/** Absent on early v1 evidence and interpreted as direct. */
|
||||
upstreamMode?: 'direct' | 'http-connect';
|
||||
proxyEnvironmentKeys: Array<
|
||||
| 'HTTP_PROXY'
|
||||
| 'HTTPS_PROXY'
|
||||
|
|
|
|||
|
|
@ -4,6 +4,8 @@ import type { TaskStatus, AgentType } from './task.types.js';
|
|||
import type { HarnessSupportFailureClass, HarnessSupportTier } from './provider-runtime.types.js';
|
||||
import type { AdmissionLaunchSource } from './admission-control.types.js';
|
||||
import type { ExecutionTreeBreakerEvidence } from './execution-tree-budget.types.js';
|
||||
import type { ExecutableAgentProvider } from './config.types.js';
|
||||
import type { RunEgressDecisionReason, RunEgressProtocol } from './sandbox-policy.types.js';
|
||||
|
||||
export type TelemetryEventType =
|
||||
| 'task.created'
|
||||
|
|
@ -14,6 +16,7 @@ export type TelemetryEventType =
|
|||
| 'run.completed'
|
||||
| 'run.error'
|
||||
| 'run.tokens'
|
||||
| 'network.egress'
|
||||
| 'admission.tree_control';
|
||||
|
||||
/** Base telemetry event - all events extend this */
|
||||
|
|
@ -125,6 +128,28 @@ export interface AdmissionTreeControlTelemetryEvent extends TelemetryEvent {
|
|||
unresolvedAttempts?: number;
|
||||
}
|
||||
|
||||
/** Metadata-only run-scoped network policy decision. */
|
||||
export interface NetworkEgressTelemetryEvent extends TelemetryEvent {
|
||||
type: 'network.egress';
|
||||
taskId: string;
|
||||
attemptId: string;
|
||||
agent: string;
|
||||
provider: ExecutableAgentProvider;
|
||||
gatewayId: string;
|
||||
runKey: string;
|
||||
policyHash: string;
|
||||
protocol: RunEgressProtocol;
|
||||
hostKey: string;
|
||||
port: number;
|
||||
method?: string;
|
||||
decision: 'allow' | 'block';
|
||||
reason: RunEgressDecisionReason;
|
||||
policyReason?: RunEgressDecisionReason;
|
||||
blockedAddressClass?: 'private' | 'loopback' | 'link-local' | 'metadata';
|
||||
approvalEligible: boolean;
|
||||
approvalId?: string;
|
||||
}
|
||||
|
||||
/** Union type for all telemetry events */
|
||||
export type AnyTelemetryEvent =
|
||||
| TaskTelemetryEvent
|
||||
|
|
@ -133,6 +158,7 @@ export type AnyTelemetryEvent =
|
|||
| RunCompletedEvent
|
||||
| RunErrorEvent
|
||||
| TokenTelemetryEvent
|
||||
| NetworkEgressTelemetryEvent
|
||||
| AdmissionTreeControlTelemetryEvent;
|
||||
|
||||
/** Telemetry configuration */
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue