-
Success Rate
+
Transfer Completion
- {getSuccessRate().toFixed(1)}%
+ {stats.total_requests ? `${getSuccessRate().toFixed(1)}%` : '—'}
@@ -245,7 +252,7 @@ export function StatisticsOverview() {
Avg Duration
-
{safeFormatDuration(stats.avg_duration_us)}
+
{stats.total_requests ? safeFormatDuration(stats.avg_duration_us) : '—'}
@@ -282,9 +289,9 @@ export function StatisticsOverview() {
-
Successful Requests
+
Completed Transfers
- {getSuccessRate().toFixed(1)}% success rate
+ {stats.total_requests ? `${getSuccessRate().toFixed(1)}%` : '—'} transfer completion
@@ -297,9 +304,9 @@ export function StatisticsOverview() {
-
Failed Requests
+
Incomplete Transfers
- {getErrorRate().toFixed(1)}% error rate
+ {stats.total_requests ? `${getErrorRate().toFixed(1)}%` : '—'} incomplete transfers
@@ -322,24 +329,24 @@ export function StatisticsOverview() {
Average Total Duration
- {safeFormatDuration(stats.avg_duration_us)}
+ {stats.total_requests ? safeFormatDuration(stats.avg_duration_us) : '—'}
-
+
- Average Upstream Latency
- {safeFormatDuration(stats.avg_upstream_latency_us)}
+ Average time to headers / attempt end
+ {stats.total_requests ? safeFormatDuration(stats.avg_upstream_latency_us) : '—'}
-
+
- Average Proxy Overhead
- {safeFormatDuration(stats.avg_proxy_overhead_us)}
+ Average other elapsed time
+ {stats.total_requests ? safeFormatDuration(stats.avg_proxy_overhead_us) : '—'}
- Proxy Overhead Percentage
+ Other elapsed percentage
- {getProxyOverheadPercentage().toFixed(1)}%
+ {stats.total_requests ? `${getProxyOverheadPercentage().toFixed(1)}%` : '—'}
@@ -362,12 +369,12 @@ export function StatisticsOverview() {
Total Request Data
{safeFormatBytes(stats.total_request_size)}
-
+
Total Response Data
{safeFormatBytes(stats.total_response_size)}
-
+
Combined Total
@@ -385,7 +392,7 @@ export function StatisticsOverview() {
- Status Codes Distribution
+ Upstream HTTP Status Distribution
@@ -450,4 +457,4 @@ export function StatisticsOverview() {
)}
);
-}
\ No newline at end of file
+}
diff --git a/dashboard/src/hooks/useRequestStore.ts b/dashboard/src/hooks/useRequestStore.ts
index 3d648e7..166c599 100644
--- a/dashboard/src/hooks/useRequestStore.ts
+++ b/dashboard/src/hooks/useRequestStore.ts
@@ -1,5 +1,4 @@
import { create } from 'zustand';
-import { persist } from 'zustand/middleware';
import { RequestConfig, ApiResponse, RequestHistory, HeaderPair, HttpMethod } from '../types/api';
interface RequestStore {
@@ -8,32 +7,32 @@ interface RequestStore {
url: string;
headers: HeaderPair[];
body: string;
-
+
// Response state
response: ApiResponse | null;
isLoading: boolean;
error: string | null;
-
+
// History
history: RequestHistory[];
-
+
// Actions
setMethod: (method: HttpMethod) => void;
setUrl: (url: string) => void;
setHeaders: (headers: HeaderPair[]) => void;
setBody: (body: string) => void;
-
+
addHeader: () => void;
removeHeader: (index: number) => void;
updateHeader: (index: number, key: string, value: string, enabled: boolean) => void;
-
+
setResponse: (response: ApiResponse | null) => void;
setLoading: (loading: boolean) => void;
setError: (error: string | null) => void;
-
+
addToHistory: (request: RequestConfig, response?: ApiResponse, error?: string) => void;
clearHistory: () => void;
-
+
reset: () => void;
}
@@ -43,79 +42,67 @@ const defaultHeaders: HeaderPair[] = [
];
export const useRequestStore = create
()(
- persist(
- (set, get) => ({
- // Initial state
+ (set, get) => ({
+ // Initial state
+ method: 'GET',
+ url: '',
+ headers: defaultHeaders,
+ body: '',
+
+ response: null,
+ isLoading: false,
+ error: null,
+
+ history: [],
+
+ // Actions
+ setMethod: (method) => set({ method }),
+ setUrl: (url) => set({ url }),
+ setHeaders: (headers) => set({ headers }),
+ setBody: (body) => set({ body }),
+
+ addHeader: () => {
+ const { headers } = get();
+ set({ headers: [...headers, { key: '', value: '', enabled: true }] });
+ },
+
+ removeHeader: (index) => {
+ const { headers } = get();
+ set({ headers: headers.filter((_, i) => i !== index) });
+ },
+
+ updateHeader: (index, key, value, enabled) => {
+ const { headers } = get();
+ const newHeaders = [...headers];
+ newHeaders[index] = { key, value, enabled };
+ set({ headers: newHeaders });
+ },
+
+ setResponse: (response) => set({ response }),
+ setLoading: (isLoading) => set({ isLoading }),
+ setError: (error) => set({ error }),
+
+ addToHistory: (request, response, error) => {
+ const { history } = get();
+ const newEntry: RequestHistory = {
+ id: Date.now().toString(),
+ request,
+ response,
+ error,
+ timestamp: Date.now(),
+ };
+ set({ history: [newEntry, ...history.slice(0, 99)] }); // Keep last 100 entries
+ },
+
+ clearHistory: () => set({ history: [] }),
+
+ reset: () => set({
method: 'GET',
url: '',
headers: defaultHeaders,
body: '',
-
response: null,
- isLoading: false,
error: null,
-
- history: [],
-
- // Actions
- setMethod: (method) => set({ method }),
- setUrl: (url) => set({ url }),
- setHeaders: (headers) => set({ headers }),
- setBody: (body) => set({ body }),
-
- addHeader: () => {
- const { headers } = get();
- set({ headers: [...headers, { key: '', value: '', enabled: true }] });
- },
-
- removeHeader: (index) => {
- const { headers } = get();
- set({ headers: headers.filter((_, i) => i !== index) });
- },
-
- updateHeader: (index, key, value, enabled) => {
- const { headers } = get();
- const newHeaders = [...headers];
- newHeaders[index] = { key, value, enabled };
- set({ headers: newHeaders });
- },
-
- setResponse: (response) => set({ response }),
- setLoading: (isLoading) => set({ isLoading }),
- setError: (error) => set({ error }),
-
- addToHistory: (request, response, error) => {
- const { history } = get();
- const newEntry: RequestHistory = {
- id: Date.now().toString(),
- request,
- response,
- error,
- timestamp: Date.now(),
- };
- set({ history: [newEntry, ...history.slice(0, 99)] }); // Keep last 100 entries
- },
-
- clearHistory: () => set({ history: [] }),
-
- reset: () => set({
- method: 'GET',
- url: '',
- headers: defaultHeaders,
- body: '',
- response: null,
- error: null,
- }),
}),
- {
- name: 'fetchr-request-store',
- partialize: (state) => ({
- method: state.method,
- url: state.url,
- headers: state.headers,
- body: state.body,
- history: state.history,
- }),
- }
- )
-);
\ No newline at end of file
+ })
+);
diff --git a/dashboard/src/lib/inspection.test.ts b/dashboard/src/lib/inspection.test.ts
new file mode 100644
index 0000000..a61b873
--- /dev/null
+++ b/dashboard/src/lib/inspection.test.ts
@@ -0,0 +1,15 @@
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { redactBody, redactHeaders, redactURL, REDACTED, OMITTED_BODY } from './inspection';
+
+test('sanitizes nested JSON, URL credentials and custom fields without hiding ordinary fields', () => {
+ const body = redactBody(JSON.stringify({ password: 'fixture-password', nested: [{ custom_pin: 'fixture-pin', n: 42 }], link: 'https://user:fixture-user@example.test/bot123:fixture-token/getMe?q=fixture-query' }), ['custom_pin']);
+ assert.equal(body.includes('fixture-'), false);
+ assert.equal(JSON.parse(body).nested[0].n, 42);
+ assert.equal(redactHeaders({ Authorization: 'fixture-auth', Cookie: 'fixture-cookie', 'X-Pin': 'fixture-pin' }, ['X-Pin'])['X-Pin'], REDACTED);
+ assert.equal(redactURL('https://example.test/token/fixture-path?q=fixture-query').includes('fixture-'), false);
+ assert.equal(redactURL('https://example.test/%62ot123%3Afixture-token/getMe').includes('fixture-'), false);
+});
+test('omits opaque, incomplete and oversized bodies', () => {
+ for (const body of ['data: fixture-token', '{"password":"fixture"', 'x'.repeat(65537)]) assert.equal(redactBody(body), OMITTED_BODY);
+});
diff --git a/dashboard/src/lib/inspection.ts b/dashboard/src/lib/inspection.ts
new file mode 100644
index 0000000..0cd5346
--- /dev/null
+++ b/dashboard/src/lib/inspection.ts
@@ -0,0 +1,47 @@
+import type { BackendRequestRecord } from '../services/api';
+
+export const REDACTED = '[REDACTED]';
+export const OMITTED_BODY = '[Body omitted: only complete JSON up to 64 KiB is retained]';
+export function sensitiveField(key: string, extra: string[] = []): boolean {
+ const normalize = (s: string) => s.toLowerCase().replace(/[-_ ]/g, '');
+ return /authorization|cookie|password|passwd|secret|token|apikey|credential|privatekey/.test(normalize(key)) || extra.some(s => normalize(s.trim()) === normalize(key));
+}
+export function redactText(value: string): string {
+ return value.replace(/bearer\s+[a-z0-9._~+/-]+=*|bot[0-9]+:[a-z0-9_-]+|\beyJ[a-z0-9_-]+\.[a-z0-9_-]+\.[a-z0-9_-]+/gi, REDACTED);
+}
+export function redactURL(value: string): string {
+ try {
+ const url = new URL(value);
+ if (url.username || url.password) { url.username = REDACTED; url.password = ''; }
+ url.hash = '';
+ for (const key of Array.from(url.searchParams.keys())) url.searchParams.set(key, REDACTED);
+ let previousSensitive = false;
+ url.pathname = url.pathname.split('/').map(encoded => {
+ const segment = decodeURIComponent(encoded);
+ const hide = previousSensitive || segment.includes(':') || segment.length > 32;
+ previousSensitive = sensitiveField(segment);
+ return encodeURIComponent(hide ? REDACTED : redactText(segment));
+ }).join('/');
+ return url.toString();
+ } catch { return REDACTED; }
+}
+export function redactHeaders(headers: Record, extra: string[] = []): Record {
+ return Object.fromEntries(Object.entries(headers).map(([key, value]) => [key,
+ sensitiveField(key, extra) ? REDACTED : /^(location|referer|x-netkit-destination)$/i.test(key) ? redactURL(value) : redactText(value),
+ ]));
+}
+export function redactBody(body: string, extra: string[] = []): string {
+ if (!body || body === OMITTED_BODY) return body;
+ if (new TextEncoder().encode(body).length > 65536) return OMITTED_BODY;
+ const visit = (value: unknown): unknown => {
+ if (Array.isArray(value)) return value.map(visit);
+ if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value).map(([key, val]) => [key, sensitiveField(key, extra) ? REDACTED : visit(val)]));
+ if (typeof value === 'string') return /^https?:\/\//.test(value) ? redactURL(value) : redactText(value);
+ return value;
+ };
+ try { return JSON.stringify(visit(JSON.parse(body))); } catch { return OMITTED_BODY; }
+}
+export function outcomeLabel(record: BackendRequestRecord): string {
+ const labels: Record = { complete: 'Transfer completed', interrupted: 'Interrupted', client_canceled: 'Client canceled', blocked_by_inspection: 'Blocked by inspection', upstream_error: 'Upstream connection failed', proxy_error: 'Proxy error', tunnel_only: 'Tunnel only', unknown: 'Unknown' };
+ return labels[record.outcome] || 'Unknown (older server)';
+}
diff --git a/dashboard/src/services/api.test.ts b/dashboard/src/services/api.test.ts
new file mode 100644
index 0000000..7f65e6b
--- /dev/null
+++ b/dashboard/src/services/api.test.ts
@@ -0,0 +1,66 @@
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { apiService } from './api';
+
+test('API failures propagate instead of becoming empty data', async () => {
+ const original = global.fetch;
+ try {
+ global.fetch = async () => new Response('unavailable', { status: 503 });
+ await assert.rejects(apiService.getRequestHistory());
+ await assert.rejects(apiService.getRequestStats());
+ global.fetch = async () => new Response('null', { status: 200 });
+ await assert.rejects(apiService.getRequestHistory());
+ await assert.rejects(apiService.getRequestStats());
+ } finally { global.fetch = original; }
+});
+
+test('older/raw history is sanitized before reaching components', async () => {
+ const original = global.fetch;
+ try {
+ global.fetch = async (url) => new Response(JSON.stringify(String(url).includes('healthz') ? { redact_fields: ['pin'] } : { records: [{
+ method: 'POST', url: 'https://example.test/bot123:fixture-token/getMe?q=fixture-query',
+ request_headers: { Authorization: 'fixture-auth' }, response_headers: { 'Set-Cookie': 'fixture-cookie' },
+ request_body: '{"pin":"fixture-pin"}', response_body: '{"password":"fixture-password"}', error: 'fixture-raw-error',
+ }] }));
+ const records = await apiService.getRequestHistory();
+ assert.equal(JSON.stringify(records).includes('fixture-'), false);
+ } finally { global.fetch = original; }
+});
+
+test('builder forwards original input but only exposes sanitized response', async () => {
+ const original = global.fetch;
+ let sent = false;
+ try {
+ global.fetch = async (url, options) => {
+ if (String(url).includes('healthz')) return new Response(JSON.stringify({ redact_fields: ['pin'] }));
+ assert.equal(new Headers(options?.headers).get('Authorization'), 'fixture-outgoing');
+ assert.equal(options?.body, '{"password":"fixture-outgoing-body"}');
+ sent = true;
+ return new Response('{"pin":"fixture-response"}', { headers: { 'X-Api-Key': 'fixture-header' } });
+ };
+ const result = await apiService.makeRequest({ method: 'POST', url: 'https://example.test', headers: { Authorization: 'fixture-outgoing' }, body: '{"password":"fixture-outgoing-body"}' });
+ assert.equal(sent, true);
+ assert.equal(JSON.stringify(result).includes('fixture-'), false);
+ } finally { global.fetch = original; }
+});
+
+test('capture policy and data endpoints preserve a configured dashboard prefix', async () => {
+ const original = global.fetch;
+ const previousWindow = Object.getOwnPropertyDescriptor(globalThis, 'window');
+ const seen: string[] = [];
+ try {
+ Object.defineProperty(globalThis, 'window', { value: { __NETKIT_CONFIG__: { basePath: '/tools/netkit' } }, configurable: true });
+ global.fetch = async (url) => {
+ seen.push(String(url));
+ return new Response(JSON.stringify(String(url).includes('healthz') ? { capture_mode: 'http_and_connect_tunnels', history_capacity: 5 } : String(url).includes('/stats') ? { total_requests: 0 } : { records: [] }));
+ };
+ await apiService.getRequestHistory();
+ await apiService.getRequestStats();
+ assert.equal(seen.length, 3);
+ assert.ok(seen.every(url => url.startsWith('/tools/netkit/api/admin/')));
+ } finally {
+ global.fetch = original;
+ if (previousWindow) Object.defineProperty(globalThis, 'window', previousWindow);
+ else Reflect.deleteProperty(globalThis, 'window');
+ }
+});
diff --git a/dashboard/src/services/api.ts b/dashboard/src/services/api.ts
index 59a3a23..b1ceb3f 100644
--- a/dashboard/src/services/api.ts
+++ b/dashboard/src/services/api.ts
@@ -1,3 +1,4 @@
+import { redactHeaders, redactBody, redactURL } from '../lib/inspection';
import { RequestConfig, ApiResponse } from '../types/api';
// Backend request record from the Go API
@@ -20,6 +21,9 @@ export interface BackendRequestRecord {
total_duration_us: number;
request_size: number;
response_size: number;
+ outcome: string;
+ response_source: string;
+ failure_reason?: string;
success: boolean;
error?: string;
}
@@ -30,6 +34,11 @@ export interface BackendHistoryResponse {
}
export interface RequestStats {
+ scope: string;
+ capacity: number;
+ oldest_at?: string;
+ newest_at?: string;
+ outcomes?: Record;
total_requests: number;
success_count: number;
error_count: number;
@@ -171,10 +180,18 @@ class ApiService {
return `${url}${separator}_t=${Date.now()}`;
}
+ async getCaptureInfo(): Promise<{ capture_mode: string; history_capacity: number; redact_fields?: string[] }> {
+ const response = await fetch(joinUrl(this.getAdminBaseUrl(), "/healthz"), { cache: "no-store" });
+ if (!response.ok) throw new Error("Capture status unavailable");
+ return response.json();
+ }
+
async makeRequest(config: RequestConfig): Promise {
const startTime = Date.now();
try {
+ // Fetch the server policy before sending; do not display an unsanitized response.
+ const policy = await this.getCaptureInfo();
// Convert headers object to fetch headers
const headers = new Headers();
@@ -225,17 +242,17 @@ class ApiService {
return {
statusCode: response.status,
- headers: responseHeaders,
- body: responseBody,
+ headers: redactHeaders(responseHeaders, policy.redact_fields),
+ body: redactBody(responseBody, policy.redact_fields),
timestamp: startTime,
duration: endTime - startTime,
};
- } catch (error: unknown) {
+ } catch {
const endTime = Date.now();
throw {
statusCode: 0,
headers: {},
- body: error instanceof Error ? error.message : 'Unknown error',
+ body: 'Request failed. Check the capture history and API connection.',
timestamp: startTime,
duration: endTime - startTime,
};
@@ -255,15 +272,23 @@ class ApiService {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
const data: BackendHistoryResponse = await response.json();
- return data.records;
+ if (!data || !Array.isArray(data.records)) throw new Error('Invalid traffic response');
+ const policy = await this.getCaptureInfo();
+ return data.records.map(record => ({ ...record,
+ url: record.method === 'CONNECT' && /^[a-z0-9.[\]:-]+$/i.test(record.url) ? record.url : redactURL(record.url),
+ request_headers: redactHeaders(record.request_headers || {}, policy.redact_fields),
+ response_headers: redactHeaders(record.response_headers || {}, policy.redact_fields),
+ request_body: redactBody(record.request_body || '', policy.redact_fields),
+ response_body: redactBody(record.response_body || '', policy.redact_fields),
+ error: record.error ? (record.outcome ? 'Exchange incomplete; see outcome and failure reason.' : 'Exchange incomplete (older server).') : undefined,
+ }));
} catch (error) {
- console.error('Failed to fetch request history:', error);
- return [];
+ throw error;
}
}
// Get request statistics from the backend
- async getRequestStats(): Promise {
+ async getRequestStats(): Promise {
try {
const url = this.addCacheBuster(joinUrl(this.getAdminBaseUrl(), '/requests/stats'));
const response = await fetch(url, {
@@ -274,10 +299,11 @@ class ApiService {
if (!response.ok) {
throw new Error(`HTTP ${response.status}: ${response.statusText}`);
}
- return await response.json();
+ const data: RequestStats = await response.json();
+ if (!data || !Number.isFinite(data.total_requests)) throw new Error('Invalid statistics response');
+ return data;
} catch (error) {
- console.error('Failed to fetch request stats:', error);
- return null;
+ throw error;
}
}
diff --git a/docs/trustworthy-capture.md b/docs/trustworthy-capture.md
new file mode 100644
index 0000000..75cc7a0
--- /dev/null
+++ b/docs/trustworthy-capture.md
@@ -0,0 +1,63 @@
+# Capture safety and signal semantics
+
+Netkit sanitizes history **before retention and admin API serialization**. Forwarded requests and responses keep their original values. The dashboard sanitizes received captures and builder responses as defense in depth; response copies use the same sanitized value.
+
+## Redaction policy
+
+- Header and JSON field names containing authorization, cookie, password, passwd, secret, token, apikey, credential, or privatekey are masked, ignoring case, hyphens, underscores and spaces.
+- `--redact-fields customer_pin,x-internal-key` adds exact normalized JSON/header field names. Matching applies recursively to JSON arrays and objects. This policy is exposed by health metadata so the builder applies it too.
+- URL userinfo and all query values are masked; fragments are removed. Credential-shaped path segments (including Telegram bot tokens), segments following sensitive field names, colon-bearing segments, and segments longer than 32 characters are masked. Location, Referer, and X-Netkit-Destination headers and URL-valued JSON strings use the same policy.
+- Bearer values, JWT-shaped strings and Telegram bot tokens are masked in retained strings.
+- Only complete JSON bodies up to 64 KiB are retained after sanitization. Non-JSON, partial JSON, SSE, binary and larger bodies have an explicit omission marker. Streaming forwarding is unaffected; inspection capture buffers are bounded while byte counts track the full observed stream.
+- Transport errors use controlled messages and bounded failure reasons (including TLS certificate and timeout), not raw exceptions containing target URLs.
+
+This is a field/pattern policy, not a general secret detector. Custom credentials in ordinary path segments, arbitrary text fields or unrecognized headers require policy changes or additional `--redact-fields`. Inspect only trusted traffic and restrict access to the proxy/admin service. There is no raw-capture reveal/export option in this release.
+
+Builder drafts remain in tab memory; no request state is written to browser storage. Existing `fetchr-request-store` saved drafts are discarded on dashboard load without hydration. A storage-access failure displays a cleanup notice. Values the user deliberately types in the composer are editable, and may be visible; this is distinct from sanitized captured evidence. Sending a draft containing redaction/omission markers is blocked until the missing values are replaced. Sanitized URLs are copied, not opened as if they were a replay.
+
+## Outcomes and timing
+
+`response_status` is accompanied by `response_source` (`upstream` or `proxy`) and `outcome`:
+
+| Outcome | Meaning |
+|---|---|
+| complete | Transfer completed, including HTTP 4xx/5xx responses. This is not HTTP success. |
+| interrupted | Transfer ended early; an already received upstream status is preserved. |
+| client_canceled | The downstream request context was canceled. |
+| blocked_by_inspection | A protocol upgrade was rejected by inspection mode; status belongs to Netkit. |
+| upstream_error | Upstream connection/request failed before response headers. |
+| proxy_error | Netkit could not process the request or establish a tunnel. |
+| tunnel_only | Opaque CONNECT tunnel established. Encrypted inner requests and later tunnel lifetime are not observed. |
+| unknown | Outcome is unavailable; do not infer success. |
+
+The legacy `success` field means a completed transfer or established tunnel. The dashboard labels it **Transfer Completion**. HTTP status distribution includes upstream statuses only; its denominator is retained observations, which can also include proxy failures and tunnels.
+
+`total_duration_us` measures the handler lifetime through forwarding completion/failure. CONNECT measures establishment only. `upstream_latency_us` is dispatch to response headers, or the end of a failed connection attempt. Missing phase timestamps yield zero rather than an overflowing duration. `proxy_overhead_us` is retained for compatibility but is labeled **Other elapsed time**: total minus the upstream phase, including body transfer and downstream waiting, not pure processing overhead. Polling and streaming can legitimately take much longer than ordinary requests.
+
+History and statistics describe only the retained in-memory buffer, with capacity and oldest/newest timestamps. They are not lifetime totals. Health metadata exposes capture mode: HTTP with opaque HTTPS tunnels, or HTTP/HTTPS inspection. HTTPS inspection cannot handle protocol upgrades; exchanges appear when completed, not while still streaming. Inspection connection setup failures before an inner HTTP request are not currently individual history records or exchange metrics.
+
+## Metrics
+
+`/metrics` (also `/api/admin/metrics` under the dashboard base path) exports:
+
+| Metric | Type | Scope |
+|---|---|---|
+| netkit_requests_total | Counter | Total observations since process start. |
+| netkit_exchanges_total | Counter | Observations by bounded method, outcome and upstream status class. |
+| netkit_exchange_duration_seconds | Histogram | Total handler lifetime / CONNECT establishment, same labels. |
+| netkit_response_headers_seconds | Histogram | Dispatch to upstream response headers, excluding opaque tunnels and failed attempts. |
+| netkit_history_records | Gauge | Current retained count. |
+
+Counters/histograms survive history eviction and clear; they reset on process restart. They do not count dashboard/admin reads, CORS preflights, or encrypted requests inside opaque tunnels. Labels never include raw hosts, paths, URLs, IDs or credentials. Nonstandard methods map to OTHER, and missing/non-upstream statuses map to none. Host/route dimensions and alert rules are later roadmap deliverables.
+
+The old constant `netkit_proxy_status` sample is removed. Use the external scraper's `up` signal for reachability; successful scraping does not prove traffic is arriving or that dependencies are healthy.
+
+## Dashboard failure states
+
+History and statistics fetch failures propagate to components rather than becoming empty arrays/null. The UI distinguishes initial loading, an empty successful capture, no filter matches, unavailable data and stale retained results after failed refresh. Last successful refresh remains visible. API reachability is distinct from traffic freshness; history refresh is manual in this release.
+
+## Validation
+
+Run `go test -race -tags=unit ./...`, `go test -race -tags=e2e ./test/e2e`, and `golangci-lint run`. In `dashboard`, run `npm run test:unit`, `npx tsc --noEmit`, `npx eslint src`, and `npm run build`.
+
+Regression fixtures cover raw forwarding versus sanitized storage/API, nested/custom fields, encoded URL credentials, incomplete/oversized payloads, complete 200/401 responses, interrupted 200, inspection rejection, concurrent scrapes, bounded labels and counters surviving eviction/clear. Browser checks should additionally exercise failure-after-success, empty capture, request detail, sanitized copy and draft reset after reload.
diff --git a/internal/proxy/history.go b/internal/proxy/history.go
index 770ba15..a82df8e 100644
--- a/internal/proxy/history.go
+++ b/internal/proxy/history.go
@@ -25,8 +25,8 @@ type RequestRecord struct {
ProxyEndTime time.Time `json:"proxy_end_time"`
// Calculated metrics (in microseconds for better precision)
- ProxyOverheadUs int64 `json:"proxy_overhead_us"` // Time spent in proxy logic (microseconds)
- UpstreamLatencyUs int64 `json:"upstream_latency_us"` // Time waiting for upstream (microseconds)
+ ProxyOverheadUs int64 `json:"proxy_overhead_us"` // Legacy residual: total minus time to headers; includes body transfer, NOT CPU overhead
+ UpstreamLatencyUs int64 `json:"upstream_latency_us"` // Time to response headers (or failed upstream attempt); not full body latency
TotalDurationUs int64 `json:"total_duration_us"` // Total time from client perspective (microseconds)
// Size metrics
@@ -34,15 +34,21 @@ type RequestRecord struct {
ResponseSize int64 `json:"response_size"`
// Status
- Success bool `json:"success"`
- Error string `json:"error,omitempty"`
+ FailureReason string `json:"failure_reason,omitempty"`
+ Outcome string `json:"outcome"`
+ ResponseSource string `json:"response_source"`
+ Success bool `json:"success"`
+ Error string `json:"error,omitempty"`
}
// RequestHistory manages the collection of request records
type RequestHistory struct {
- records []RequestRecord
- mutex sync.RWMutex
- maxSize int
+ records []RequestRecord
+ mutex sync.RWMutex
+ maxSize int
+ redactFields []string
+ signals map[string]*signalSeries
+ observed uint64
}
// NewRequestHistory creates a new request history with the specified maximum size
@@ -50,6 +56,7 @@ func NewRequestHistory(maxSize int) *RequestHistory {
return &RequestHistory{
records: make([]RequestRecord, 0),
maxSize: maxSize,
+ signals: make(map[string]*signalSeries),
}
}
@@ -58,10 +65,9 @@ func (h *RequestHistory) AddRecord(record RequestRecord) {
h.mutex.Lock()
defer h.mutex.Unlock()
- // Calculate metrics
- record.TotalDurationUs = record.ProxyEndTime.Sub(record.ProxyStartTime).Microseconds()
- record.UpstreamLatencyUs = record.UpstreamEndTime.Sub(record.UpstreamStartTime).Microseconds()
- record.ProxyOverheadUs = record.TotalDurationUs - record.UpstreamLatencyUs
+ normalizeRecord(&record)
+ h.observe(record)
+ record = sanitizeRecord(record, h.redactFields)
// Add to beginning of slice (most recent first)
h.records = append([]RequestRecord{record}, h.records...)
@@ -79,7 +85,11 @@ func (h *RequestHistory) GetRecords() []RequestRecord {
// Return a copy to avoid race conditions
result := make([]RequestRecord, len(h.records))
- copy(result, h.records)
+ for i, record := range h.records {
+ result[i] = record
+ result[i].RequestHeaders = redactHeaders(record.RequestHeaders, h.redactFields)
+ result[i].ResponseHeaders = redactHeaders(record.ResponseHeaders, h.redactFields)
+ }
return result
}
@@ -87,8 +97,10 @@ func (h *RequestHistory) GetRecords() []RequestRecord {
func (h *RequestHistory) GetRecordsJSON() ([]byte, error) {
records := h.GetRecords()
return json.Marshal(map[string]interface{}{
- "records": records,
- "total": len(records),
+ "records": records,
+ "total": len(records),
+ "capacity": h.maxSize,
+ "scope": "retained_buffer",
})
}
@@ -106,7 +118,10 @@ func (h *RequestHistory) GetStats() map[string]interface{} {
if len(h.records) == 0 {
return map[string]interface{}{
- "total_requests": 0,
+ "total_requests": 0, "success_count": 0, "error_count": 0,
+ "avg_duration_us": 0, "avg_upstream_latency_us": 0, "avg_proxy_overhead_us": 0,
+ "total_request_size": 0, "total_response_size": 0,
+ "scope": "retained_buffer", "capacity": h.maxSize,
}
}
@@ -115,8 +130,17 @@ func (h *RequestHistory) GetStats() map[string]interface{} {
var successCount, errorCount int
statusCounts := make(map[int]int)
methodCounts := make(map[string]int)
+ outcomeCounts := make(map[string]int)
+ var oldest, newest time.Time
for _, record := range h.records {
+ outcomeCounts[record.Outcome]++
+ if oldest.IsZero() || record.Timestamp.Before(oldest) {
+ oldest = record.Timestamp
+ }
+ if record.Timestamp.After(newest) {
+ newest = record.Timestamp
+ }
totalDuration += record.TotalDurationUs
totalUpstreamLatency += record.UpstreamLatencyUs
totalProxyOverhead += record.ProxyOverheadUs
@@ -129,13 +153,17 @@ func (h *RequestHistory) GetStats() map[string]interface{} {
errorCount++
}
- statusCounts[record.ResponseStatus]++
+ if record.ResponseSource == "upstream" && record.ResponseStatus > 0 {
+ statusCounts[record.ResponseStatus]++
+ }
methodCounts[record.Method]++
}
count := len(h.records)
return map[string]interface{}{
- "total_requests": count,
+ "total_requests": count,
+ "outcomes": outcomeCounts, "scope": "retained_buffer", "capacity": h.maxSize,
+ "oldest_at": oldest, "newest_at": newest,
"success_count": successCount,
"error_count": errorCount,
"avg_duration_us": totalDuration / int64(count),
diff --git a/internal/proxy/inspection.go b/internal/proxy/inspection.go
index 496bf7c..4623d2a 100644
--- a/internal/proxy/inspection.go
+++ b/internal/proxy/inspection.go
@@ -125,19 +125,24 @@ func (p *Proxy) handleInspectedConnect(w http.ResponseWriter, r *http.Request) {
// is forwarded immediately (including SSE), and the completed exchange is saved
// when the handler returns. Upload reads may run in a transport goroutine.
type synchronizedCapture struct {
- mu sync.Mutex
- data bytes.Buffer
+ mu sync.Mutex
+ data bytes.Buffer
+ total int64
}
func (c *synchronizedCapture) Write(b []byte) (int, error) {
c.mu.Lock()
defer c.mu.Unlock()
- return c.data.Write(b)
+ c.total += int64(len(b))
+ if remaining := maxCaptureBytes + 1 - c.data.Len(); remaining > 0 {
+ _, _ = c.data.Write(b[:min(len(b), remaining)])
+ }
+ return len(b), nil
}
func (c *synchronizedCapture) snapshot() (string, int64) {
c.mu.Lock()
defer c.mu.Unlock()
- return c.data.String(), int64(c.data.Len())
+ return c.data.String(), c.total
}
type captureReadCloser struct {
@@ -161,6 +166,8 @@ func (p *Proxy) inspectHTTP(w http.ResponseWriter, r *http.Request, scheme, auth
}()
if r.Header.Get("Upgrade") != "" {
record.Error = "Protocol upgrades are not supported in inspection mode"
+ record.Outcome = "blocked_by_inspection"
+ record.ResponseSource = "proxy"
record.ResponseStatus = http.StatusNotImplemented
http.Error(w, record.Error, record.ResponseStatus)
return
@@ -184,12 +191,20 @@ func (p *Proxy) inspectHTTP(w http.ResponseWriter, r *http.Request, scheme, auth
record.ResponseStatus = resp.StatusCode
record.ResponseHeaders = convertHeaders(resp.Header)
record.Success = true
+ record.Outcome = "complete"
+ record.ResponseSource = "upstream"
resp.Body = &captureReadCloser{Reader: io.TeeReader(resp.Body, &responseBody), Closer: resp.Body}
return nil
},
ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
record.UpstreamEndTime = time.Now()
- record.Error = err.Error()
+ record.Error = "Upstream request failed"
+ record.Outcome = "upstream_error"
+ record.FailureReason = failureReason(err)
+ if r.Context().Err() != nil {
+ record.Outcome = "client_canceled"
+ }
+ record.ResponseSource = "proxy"
record.ResponseStatus = http.StatusBadGateway
http.Error(w, "Upstream request failed", http.StatusBadGateway)
},
@@ -200,6 +215,10 @@ func (p *Proxy) inspectHTTP(w http.ResponseWriter, r *http.Request, scheme, auth
if v := recover(); v != nil {
record.Success = false
record.Error = "Response stream interrupted"
+ record.Outcome = "interrupted"
+ if r.Context().Err() != nil {
+ record.Outcome = "client_canceled"
+ }
panic(v)
}
}()
diff --git a/internal/proxy/inspection_test.go b/internal/proxy/inspection_test.go
index 6998005..ceadbe2 100644
--- a/internal/proxy/inspection_test.go
+++ b/internal/proxy/inspection_test.go
@@ -83,9 +83,9 @@ func TestInspectionCapturesHTTPSBodiesAndReusesConnection(t *testing.T) {
require.Len(t, records, 2)
for _, r := range records {
require.Equal(t, "POST", r.Method)
- require.Equal(t, upstream.URL+"/body?q=test", r.URL)
+ require.Equal(t, redactURL(upstream.URL+"/body?q=test"), r.URL)
require.Equal(t, `{"hello":"world"}`, r.RequestBody)
- require.Equal(t, `response:{"hello":"world"}`, r.ResponseBody)
+ require.Equal(t, omittedBody, r.ResponseBody)
require.Equal(t, 201, r.ResponseStatus)
require.True(t, r.Success)
}
@@ -114,7 +114,7 @@ func TestInspectionStreamsBeforeResponseCompletes(t *testing.T) {
// Closing the downstream stream must cancel the upstream and record capture.
require.NoError(t, resp.Body.Close())
r := waitRecords(t, p, 1)[0]
- require.Contains(t, r.ResponseBody, "data: first")
+ require.Equal(t, omittedBody, r.ResponseBody)
}
func TestInspectionVerifiesOriginCertificate(t *testing.T) {
@@ -172,7 +172,7 @@ func TestInspectionPlaintextCONNECT(t *testing.T) {
require.Equal(t, "plain-body", string(b))
r := waitRecords(t, p, 1)[0]
require.Equal(t, plain.URL+"/plain", r.URL)
- require.Equal(t, "plain-body", r.ResponseBody)
+ require.Equal(t, omittedBody, r.ResponseBody)
}
func TestInspectionPreservesRedirectsAndOptions(t *testing.T) {
@@ -264,7 +264,7 @@ func TestInspectionLongPolling(t *testing.T) {
record := waitRecords(t, p, 1)[0]
if tc.status == 200 {
require.Equal(t, "poll-result", string(body))
- require.Equal(t, "poll-result", record.ResponseBody)
+ require.Equal(t, omittedBody, record.ResponseBody)
} else {
require.False(t, record.Success)
require.Contains(t, record.Error, "timeout")
diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go
index fca9326..91ec165 100644
--- a/internal/proxy/proxy.go
+++ b/internal/proxy/proxy.go
@@ -21,6 +21,7 @@ import (
// Config holds the proxy configuration
type Config struct {
+ RedactFields []string // Additional case-insensitive JSON/header field names to redact
InspectionResponseHeaderTimeout time.Duration // Zero allows long polling without a proxy header deadline
InspectionCA *CertificateAuthority // Non-nil enables inspection for every CONNECT destination
Port int
@@ -63,6 +64,7 @@ func New(config *Config) *Proxy {
history: NewRequestHistory(historySize),
}
+ proxy.history.redactFields = append([]string(nil), config.RedactFields...)
proxy.inspectionConns = make(map[net.Conn]struct{})
proxy.inspectionTransport = http.DefaultTransport.(*http.Transport).Clone()
proxy.inspectionTransport.ResponseHeaderTimeout = config.InspectionResponseHeaderTimeout
@@ -208,7 +210,7 @@ func joinDashboardPath(basePath, routePath string) string {
func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// Debug logging for received requests
if p.config.LogLevel == "debug" {
- log.Printf("Received request: %s %s", r.Method, r.URL.String())
+ log.Printf("Received request: %s %s", metricMethod(r.Method), redactURL(r.URL.String()))
}
// For CONNECT method (HTTPS tunneling)
@@ -259,6 +261,7 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
RequestSize: requestSize,
ProxyStartTime: proxyStartTime,
Success: false, // Will be updated based on outcome
+ ResponseSource: "proxy",
}
// Check for X-Netkit-Destination header (for dashboard requests)
@@ -270,6 +273,7 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
targetURL, err = url.Parse(destinationHeader)
if err != nil {
record.Error = "Invalid X-Netkit-Destination URL"
+ record.ResponseStatus = http.StatusBadRequest
record.ProxyEndTime = time.Now()
p.history.AddRecord(record)
http.Error(w, "Invalid X-Netkit-Destination URL", http.StatusBadRequest)
@@ -282,6 +286,7 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
targetURL, err = url.Parse(r.URL.String())
if err != nil {
record.Error = "Invalid URL"
+ record.ResponseStatus = http.StatusBadRequest
record.ProxyEndTime = time.Now()
p.history.AddRecord(record)
http.Error(w, "Invalid URL", http.StatusBadRequest)
@@ -290,9 +295,10 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
}
// Create the proxied request
- proxyReq, err := http.NewRequest(r.Method, targetURL.String(), bodyReader)
+ proxyReq, err := http.NewRequestWithContext(r.Context(), r.Method, targetURL.String(), bodyReader)
if err != nil {
record.Error = "Failed to create proxy request"
+ record.ResponseStatus = http.StatusInternalServerError
record.ProxyEndTime = time.Now()
p.history.AddRecord(record)
http.Error(w, "Failed to create proxy request", http.StatusInternalServerError)
@@ -317,6 +323,12 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
if err != nil {
record.Error = "Failed to proxy request"
+ record.ResponseStatus = http.StatusBadGateway
+ record.Outcome = "upstream_error"
+ record.FailureReason = failureReason(err)
+ if r.Context().Err() != nil {
+ record.Outcome = "client_canceled"
+ }
record.ProxyEndTime = time.Now()
p.history.AddRecord(record)
http.Error(w, "Failed to proxy request", http.StatusBadGateway)
@@ -329,9 +341,16 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
}()
// Capture response data
+ record.ResponseStatus = resp.StatusCode
+ record.ResponseSource = "upstream"
+ record.ResponseHeaders = convertHeaders(resp.Header)
responseBody, responseSize, err := captureResponseBody(resp)
if err != nil {
record.Error = "Failed to read response body"
+ record.Outcome = "interrupted"
+ if r.Context().Err() != nil {
+ record.Outcome = "client_canceled"
+ }
record.ProxyEndTime = time.Now()
p.history.AddRecord(record)
http.Error(w, "Failed to read response body", http.StatusInternalServerError)
@@ -344,6 +363,7 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
record.ResponseBody = responseBody
record.ResponseSize = responseSize
record.Success = true
+ record.Outcome = "complete"
// End proxy processing timing here - before we start writing response to client
record.ProxyEndTime = time.Now()
@@ -375,16 +395,21 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) {
if _, err := io.Copy(w, resp.Body); err != nil {
log.Printf("Error copying response body: %v", err)
record.Error = "Failed to copy response body"
+ record.Outcome = "interrupted"
+ if r.Context().Err() != nil {
+ record.Outcome = "client_canceled"
+ }
record.Success = false
}
- // Record the request (proxy processing complete)
+ // Record the complete handler lifetime, including downstream transfer.
+ record.ProxyEndTime = time.Now()
p.history.AddRecord(record)
// Debug logging for completed requests
if p.config.LogLevel == "debug" {
log.Printf("HTTP request completed: %s %s -> %d (%dus)",
- r.Method, r.URL.String(), resp.StatusCode, record.TotalDurationUs)
+ metricMethod(r.Method), redactURL(r.URL.String()), resp.StatusCode, elapsedUS(record.ProxyStartTime, record.ProxyEndTime))
}
}
@@ -400,9 +425,12 @@ func (p *Proxy) handleConnect(w http.ResponseWriter, r *http.Request) {
}
recordResult := func(status int, err error) {
record.ResponseStatus = status
+ record.ResponseSource = "proxy"
+ record.Outcome = "tunnel_only"
record.Success = err == nil
if err != nil {
- record.Error = err.Error()
+ record.Error = "Tunnel establishment failed"
+ record.Outcome = "proxy_error"
}
record.ProxyEndTime = time.Now()
if record.UpstreamEndTime.IsZero() {
@@ -476,11 +504,22 @@ func (p *Proxy) handleHealth(w http.ResponseWriter, r *http.Request) {
return
}
+ w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
- if _, err := w.Write([]byte(`{"status":"healthy","proxy":"netkit"}`)); err != nil {
+ mode := "http_and_connect_tunnels"
+ if p.config.InspectionCA != nil {
+ mode = "http_and_https_inspection"
+ }
+ if err := json.NewEncoder(w).Encode(map[string]interface{}{
+ "status": "healthy", "proxy": "netkit", "capture_mode": mode,
+ "history_capacity": p.history.maxSize, "history_scope": "retained_buffer",
+ "redaction": "structured_json", "redact_fields": p.config.RedactFields,
+ "body_limit_bytes": maxCaptureBytes, "metrics_scope": "process_lifetime",
+ }); err != nil {
log.Printf("Error writing health response: %v", err)
}
+
}
// handleMetrics handles metrics requests
@@ -496,17 +535,9 @@ func (p *Proxy) handleMetrics(w http.ResponseWriter, r *http.Request) {
return
}
- w.Header().Set("Content-Type", "text/plain")
+ w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
w.WriteHeader(http.StatusOK)
- // Simple metrics for now - can be expanded later
- metrics := `# HELP netkit_requests_total Total number of requests handled
-# TYPE netkit_requests_total counter
-netkit_requests_total 0
-
-# HELP netkit_proxy_status Status of the proxy server
-# TYPE netkit_proxy_status gauge
-netkit_proxy_status 1
-`
+ metrics := p.history.Metrics()
if _, err := w.Write([]byte(metrics)); err != nil {
log.Printf("Error writing metrics response: %v", err)
}
@@ -536,6 +567,7 @@ func (p *Proxy) handleRequestHistory(w http.ResponseWriter, r *http.Request) {
return
}
+ w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
if _, err := w.Write(data); err != nil {
@@ -568,6 +600,7 @@ func (p *Proxy) handleRequestStats(w http.ResponseWriter, r *http.Request) {
return
}
+ w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
if _, err := w.Write(data); err != nil {
@@ -595,6 +628,7 @@ func (p *Proxy) handleClearHistory(w http.ResponseWriter, r *http.Request) {
p.history.Clear()
+ w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
if _, err := w.Write([]byte(`{"success": true, "message": "Request history cleared"}`)); err != nil {
diff --git a/internal/proxy/redaction.go b/internal/proxy/redaction.go
new file mode 100644
index 0000000..9b9629c
--- /dev/null
+++ b/internal/proxy/redaction.go
@@ -0,0 +1,148 @@
+package proxy
+
+import (
+ "encoding/json"
+ "net"
+ "net/url"
+ "regexp"
+ "strings"
+)
+
+const redacted = "[REDACTED]"
+const omittedBody = "[Body omitted: only complete JSON up to 64 KiB is retained]"
+const maxCaptureBytes = 64 * 1024
+
+var credentialPattern = regexp.MustCompile(`(?i)(bearer\s+)[a-z0-9._~+/-]+=*|bot[0-9]+:[a-z0-9_-]+|\beyJ[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+`)
+
+func sensitiveField(key string, extra []string) bool {
+ key = strings.ToLower(strings.NewReplacer("-", "", "_", "", " ", "").Replace(key))
+ for _, part := range []string{"authorization", "cookie", "password", "passwd", "secret", "token", "apikey", "credential", "privatekey"} {
+ if strings.Contains(key, part) {
+ return true
+ }
+ }
+ for _, field := range extra {
+ if key == strings.ToLower(strings.NewReplacer("-", "", "_", "", " ", "").Replace(strings.TrimSpace(field))) {
+ return true
+ }
+ }
+ return false
+}
+
+func redactText(value string) string { return credentialPattern.ReplaceAllString(value, redacted) }
+
+func redactURL(value string) string {
+ if host, port, err := net.SplitHostPort(value); err == nil && !strings.ContainsAny(value, "/@?#") {
+ return net.JoinHostPort(host, port)
+ }
+ u, err := url.Parse(value)
+ if err != nil {
+ return redacted
+ }
+ if u.User != nil {
+ u.User = url.User(redacted)
+ }
+ u.Fragment = ""
+ // Every query value is private by default, including keys unknown to Netkit.
+ if u.RawQuery != "" {
+ q, err := url.ParseQuery(u.RawQuery)
+ if err != nil {
+ u.RawQuery = "redacted"
+ } else {
+ for k := range q {
+ q[k] = []string{redacted}
+ }
+ u.RawQuery = q.Encode()
+ }
+ }
+ segments := strings.Split(u.Path, "/")
+ previousSensitive := false
+ for i, segment := range segments {
+ sensitive := sensitiveField(segment, nil)
+ if previousSensitive || strings.Contains(segment, ":") || len(segment) > 32 {
+ segments[i] = redacted
+ } else {
+ segments[i] = redactText(segment)
+ }
+ previousSensitive = sensitive
+ }
+ u.Path, u.RawPath = strings.Join(segments, "/"), ""
+ return u.String()
+}
+
+func redactHeaders(headers map[string]string, fields []string) map[string]string {
+ result := make(map[string]string, len(headers))
+ for key, value := range headers {
+ switch {
+ case sensitiveField(key, fields):
+ result[key] = redacted
+ case strings.EqualFold(key, "X-Netkit-Destination"), strings.EqualFold(key, "Location"), strings.EqualFold(key, "Referer"):
+ result[key] = redactURL(value)
+ default:
+ result[key] = redactText(value)
+ }
+ }
+ return result
+}
+
+func redactBody(body string, fields []string) string {
+ if body == "" {
+ return ""
+ }
+ if len(body) > maxCaptureBytes {
+ return omittedBody
+ }
+ var data interface{}
+ decoder := json.NewDecoder(strings.NewReader(body))
+ decoder.UseNumber()
+ if !json.Valid([]byte(body)) || decoder.Decode(&data) != nil {
+ return omittedBody
+ }
+ var visit func(interface{}) interface{}
+ visit = func(value interface{}) interface{} {
+ switch v := value.(type) {
+ case map[string]interface{}:
+ for k, item := range v {
+ if sensitiveField(k, fields) {
+ v[k] = redacted
+ } else {
+ v[k] = visit(item)
+ }
+ }
+ case []interface{}:
+ for i, item := range v {
+ v[i] = visit(item)
+ }
+ case string:
+ if strings.HasPrefix(v, "http://") || strings.HasPrefix(v, "https://") {
+ return redactURL(v)
+ }
+ return redactText(v)
+ }
+ return value
+ }
+ encoded, err := json.Marshal(visit(data))
+ if err != nil {
+ return omittedBody
+ }
+ return string(encoded)
+}
+
+func sanitizeRecord(record RequestRecord, fields []string) RequestRecord {
+ record.URL = redactURL(record.URL)
+ record.RequestHeaders = redactHeaders(record.RequestHeaders, fields)
+ record.ResponseHeaders = redactHeaders(record.ResponseHeaders, fields)
+ record.RequestBody = redactBody(record.RequestBody, fields)
+ record.ResponseBody = redactBody(record.ResponseBody, fields)
+ // Errors are controlled messages; transport errors can embed raw target URLs.
+ if record.Error != "" {
+ record.Error = outcomeDescription(record.Outcome)
+ if record.FailureReason == "tls_certificate" {
+ record.Error += ": origin certificate verification failed"
+ }
+ if record.FailureReason == "timeout" {
+ record.Error += ": timeout"
+ }
+ }
+ return record
+}
diff --git a/internal/proxy/signals.go b/internal/proxy/signals.go
new file mode 100644
index 0000000..f6bb38e
--- /dev/null
+++ b/internal/proxy/signals.go
@@ -0,0 +1,167 @@
+package proxy
+
+import (
+ "crypto/x509"
+ "errors"
+ "fmt"
+ "net"
+ "sort"
+ "strings"
+ "time"
+)
+
+func outcomeDescription(outcome string) string {
+ switch outcome {
+ case "interrupted":
+ return "Response transfer interrupted; HTTP status may already have been received"
+ case "client_canceled":
+ return "Client canceled the exchange"
+ case "blocked_by_inspection":
+ return "Protocol upgrades are not supported in inspection mode"
+ case "upstream_error":
+ return "Upstream connection or request failed"
+ case "proxy_error":
+ return "Proxy could not process the request"
+ default:
+ return "Exchange did not complete"
+ }
+}
+
+func elapsedUS(start, end time.Time) int64 {
+ if start.IsZero() || end.IsZero() || end.Before(start) {
+ return 0
+ }
+ return end.Sub(start).Microseconds()
+}
+
+func normalizeRecord(r *RequestRecord) {
+ r.TotalDurationUs = elapsedUS(r.ProxyStartTime, r.ProxyEndTime)
+ r.UpstreamLatencyUs = elapsedUS(r.UpstreamStartTime, r.UpstreamEndTime)
+ r.ProxyOverheadUs = max(int64(0), r.TotalDurationUs-r.UpstreamLatencyUs)
+ if r.ResponseSource == "" && r.ResponseStatus != 0 {
+ r.ResponseSource = "upstream"
+ }
+ if r.Outcome == "" {
+ switch {
+ case r.Method == "CONNECT" && r.Success:
+ r.Outcome = "tunnel_only"
+ case r.Success:
+ r.Outcome = "complete"
+ default:
+ r.Outcome = "proxy_error"
+ }
+ }
+ switch r.Outcome {
+ case "complete", "tunnel_only", "interrupted", "client_canceled", "blocked_by_inspection", "upstream_error", "proxy_error":
+ default:
+ r.Outcome = "unknown"
+ }
+ r.Success = r.Outcome == "complete" || r.Outcome == "tunnel_only"
+}
+
+// No user-controlled values (host, route, IDs, URLs) become metric labels.
+func metricMethod(method string) string {
+ switch method {
+ case "GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "CONNECT":
+ return method
+ }
+ return "OTHER"
+}
+func statusClass(r RequestRecord) string {
+ if r.ResponseSource != "upstream" || r.ResponseStatus < 100 || r.ResponseStatus > 599 {
+ return "none"
+ }
+ return fmt.Sprintf("%dxx", r.ResponseStatus/100)
+}
+
+var durationBounds = []float64{.001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10, 30, 60}
+
+type signalSeries struct {
+ HeaderCount uint64
+ HeaderSum float64
+ HeaderBuckets [14]uint64
+ Count uint64
+ Sum float64
+ Buckets [14]uint64
+}
+
+func (h *RequestHistory) observe(r RequestRecord) {
+ labels := fmt.Sprintf(`method=%q,outcome=%q,http_status_class=%q`, metricMethod(r.Method), r.Outcome, statusClass(r))
+ series := h.signals[labels]
+ if series == nil {
+ series = &signalSeries{}
+ h.signals[labels] = series
+ }
+ series.Count++
+ seconds := float64(r.TotalDurationUs) / 1e6
+ series.Sum += seconds
+ for i, bound := range durationBounds {
+ if seconds <= bound {
+ series.Buckets[i]++
+ }
+ }
+ if r.ResponseSource == "upstream" && !r.UpstreamStartTime.IsZero() && !r.UpstreamEndTime.IsZero() {
+ seconds := float64(r.UpstreamLatencyUs) / 1e6
+ series.HeaderCount++
+ series.HeaderSum += seconds
+ for i, bound := range durationBounds {
+ if seconds <= bound {
+ series.HeaderBuckets[i]++
+ }
+ }
+ }
+ h.observed++
+}
+
+// Metrics is cumulative since process start. Clear/eviction affect only history.
+func (h *RequestHistory) Metrics() string {
+ h.mutex.RLock()
+ defer h.mutex.RUnlock()
+ var out strings.Builder
+ fmt.Fprintf(&out, "# HELP netkit_requests_total Completed observations since process start (includes failed exchanges and tunnel establishment).\n# TYPE netkit_requests_total counter\nnetkit_requests_total %d\n", h.observed)
+ out.WriteString("# HELP netkit_exchanges_total Observations by bounded method, outcome, and upstream HTTP status class.\n# TYPE netkit_exchanges_total counter\n")
+ keys := make([]string, 0, len(h.signals))
+ for key := range h.signals {
+ keys = append(keys, key)
+ }
+ sort.Strings(keys)
+ for _, key := range keys {
+ fmt.Fprintf(&out, "netkit_exchanges_total{%s} %d\n", key, h.signals[key].Count)
+ }
+ out.WriteString("# HELP netkit_exchange_duration_seconds Handler lifetime; CONNECT measures tunnel establishment, not encrypted requests. Includes streaming and polling lifetime.\n# TYPE netkit_exchange_duration_seconds histogram\n")
+ for _, key := range keys {
+ series := h.signals[key]
+ for i, bound := range durationBounds {
+ fmt.Fprintf(&out, "netkit_exchange_duration_seconds_bucket{%s,le=%q} %d\n", key, fmt.Sprint(bound), series.Buckets[i])
+ }
+ fmt.Fprintf(&out, "netkit_exchange_duration_seconds_bucket{%s,le=\"+Inf\"} %d\nnetkit_exchange_duration_seconds_sum{%s} %g\nnetkit_exchange_duration_seconds_count{%s} %d\n", key, series.Count, key, series.Sum, key, series.Count)
+ }
+ out.WriteString("# HELP netkit_response_headers_seconds Time from dispatch to upstream response headers; excludes failed attempts and opaque tunnels.\n# TYPE netkit_response_headers_seconds histogram\n")
+ for _, key := range keys {
+ series := h.signals[key]
+ if series.HeaderCount == 0 {
+ continue
+ }
+ for i, bound := range durationBounds {
+ fmt.Fprintf(&out, "netkit_response_headers_seconds_bucket{%s,le=%q} %d\n", key, fmt.Sprint(bound), series.HeaderBuckets[i])
+ }
+ fmt.Fprintf(&out, "netkit_response_headers_seconds_bucket{%s,le=\"+Inf\"} %d\nnetkit_response_headers_seconds_sum{%s} %g\nnetkit_response_headers_seconds_count{%s} %d\n", key, series.HeaderCount, key, series.HeaderSum, key, series.HeaderCount)
+ }
+ fmt.Fprintf(&out, "# HELP netkit_history_records Retained records (not a lifetime counter).\n# TYPE netkit_history_records gauge\nnetkit_history_records %d\n", len(h.records))
+ return out.String()
+}
+
+// Keep an actionable, bounded reason without retaining errors containing raw URLs.
+func failureReason(err error) string {
+ var cert x509.UnknownAuthorityError
+ var host x509.HostnameError
+ var invalid x509.CertificateInvalidError
+ if errors.As(err, &cert) || errors.As(err, &host) || errors.As(err, &invalid) {
+ return "tls_certificate"
+ }
+ var timeout net.Error
+ if errors.As(err, &timeout) && timeout.Timeout() {
+ return "timeout"
+ }
+ return "connection"
+}
diff --git a/internal/proxy/signals_test.go b/internal/proxy/signals_test.go
new file mode 100644
index 0000000..11ba338
--- /dev/null
+++ b/internal/proxy/signals_test.go
@@ -0,0 +1,186 @@
+package proxy
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestCaptureRedactionBeforeStorage(t *testing.T) {
+ h := NewRequestHistory(2)
+ h.redactFields = []string{"custom_pin", "X-Custom-Pin"}
+ original := map[string]string{"Authorization": "Bearer fixture-authorization", "Cookie": "session=fixture-cookie", "X-Custom-Pin": "fixture-pin", "Location": "https://user:fixture-userinfo@example.test/token/fixture-path?q=fixture-query", "X-Netkit-Destination": "https://example.test/bot123:fixture-bot/getUpdates"}
+ h.AddRecord(RequestRecord{URL: "https://user:fixture-userinfo@example.test/bot123:fixture-bot/getUpdates?q=fixture-query#fixture-fragment", RequestHeaders: original, ResponseHeaders: original, RequestBody: `{"nested":[{"api_key":"fixture-key","custom_pin":"fixture-body-pin","ok":42}],"access_token":"fixture-access","link":"https://example.test/?q=fixture-json-query"}`, ResponseBody: "opaque fixture-response", Success: true})
+ stored, err := json.Marshal(h.records)
+ require.NoError(t, err)
+ for _, secret := range []string{"fixture-authorization", "fixture-cookie", "fixture-pin", "fixture-userinfo", "fixture-path", "fixture-query", "fixture-bot", "fixture-fragment", "fixture-key", "fixture-body-pin", "fixture-access", "fixture-json-query", "fixture-response"} {
+ require.NotContains(t, string(stored), secret)
+ }
+ require.Contains(t, h.records[0].RequestBody, `"ok":42`)
+ require.Equal(t, omittedBody, h.records[0].ResponseBody)
+ require.Equal(t, "Bearer fixture-authorization", original["Authorization"], "forwarded headers must not be mutated")
+ records := h.GetRecords()
+ records[0].RequestHeaders["Authorization"] = "modified"
+ require.Equal(t, redacted, h.GetRecords()[0].RequestHeaders["Authorization"])
+}
+
+func TestSanitizationFormats(t *testing.T) {
+ for _, input := range []string{`{"password":"fixture"`, "password=fixture", "data: fixture", strings.Repeat("x", maxCaptureBytes+1)} {
+ require.Equal(t, omittedBody, redactBody(input, nil))
+ }
+ for _, raw := range []string{"https://example.test/%62ot123%3Afixture-token/getMe", "https://example.test/api_key/fixture-key", "https://example.test/?q=%zz"} {
+ require.NotContains(t, redactURL(raw), "fixture")
+ }
+ require.Equal(t, `{"n":9007199254740993}`, redactBody(`{"n":9007199254740993}`, nil), "do not corrupt JSON integers")
+}
+
+func TestMetricsSurviveEvictionClearAndConcurrentAccess(t *testing.T) {
+ h := NewRequestHistory(2)
+ start := time.Now()
+ var wg sync.WaitGroup
+ for i := 0; i < 30; i++ {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ h.AddRecord(RequestRecord{Method: fmt.Sprintf("secret-method-%d", i), URL: "https://private.test/secret", Outcome: "complete", ResponseStatus: 200, ProxyStartTime: start, ProxyEndTime: start.Add(250 * time.Millisecond)})
+ _ = h.Metrics()
+ _ = h.GetRecords()
+ }(i)
+ }
+ wg.Wait()
+ require.Len(t, h.GetRecords(), 2)
+ before := h.Metrics()
+ require.Contains(t, before, "netkit_requests_total 30\n")
+ require.Contains(t, before, `method="OTHER",outcome="complete",http_status_class="2xx"`)
+ require.Contains(t, before, `le="0.25"} 30`)
+ require.Contains(t, before, `le="0.1"} 0`)
+ require.Contains(t, before, "} 7.5\n")
+ require.NotContains(t, before, "secret")
+ h.Clear()
+ require.Contains(t, h.Metrics(), "netkit_requests_total 30\n")
+ require.Contains(t, h.Metrics(), "netkit_history_records 0\n")
+ require.Len(t, h.signals, 1, "untrusted methods must not grow label cardinality")
+}
+
+func TestMissingTimingAndOutcomeScope(t *testing.T) {
+ h := NewRequestHistory(10)
+ h.AddRecord(RequestRecord{Outcome: "blocked_by_inspection", ResponseStatus: 501, ResponseSource: "proxy", Error: "raw fixture-secret", ProxyStartTime: time.Now(), ProxyEndTime: time.Now()})
+ record := h.GetRecords()[0]
+ require.Zero(t, record.UpstreamLatencyUs)
+ require.GreaterOrEqual(t, record.ProxyOverheadUs, int64(0))
+ require.Equal(t, "Protocol upgrades are not supported in inspection mode", record.Error)
+ require.Empty(t, h.GetStats()["status_codes"])
+ require.Contains(t, h.Metrics(), `http_status_class="none"`)
+ require.Equal(t, "retained_buffer", h.GetStats()["scope"])
+}
+
+func TestHTTPOutcomesAndSanitizedAdminAPI(t *testing.T) {
+ origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ require.Equal(t, "fixture-forwarded", r.Header.Get("Authorization"))
+ body, err := io.ReadAll(r.Body)
+ require.NoError(t, err)
+ require.Contains(t, string(body), "fixture-body")
+ w.Header().Set("Content-Type", "application/json")
+ switch r.URL.Path {
+ case "/unauthorized":
+ w.WriteHeader(401)
+ case "/interrupted":
+ w.Header().Set("Content-Length", "999")
+ }
+ _, err = io.WriteString(w, `{"token":"fixture-response","ok":true}`)
+ require.NoError(t, err)
+ }))
+ defer origin.Close()
+ p := New(&Config{})
+ for _, tc := range []struct {
+ path, outcome string
+ status int
+ }{{"/ok", "complete", 200}, {"/unauthorized", "complete", 401}, {"/interrupted", "interrupted", 200}} {
+ t.Run(tc.path, func(t *testing.T) {
+ req := httptest.NewRequest("POST", origin.URL+tc.path+"?key=fixture-query", strings.NewReader(`{"password":"fixture-body"}`))
+ req.Header.Set("Authorization", "fixture-forwarded")
+ response := httptest.NewRecorder()
+ p.ServeHTTP(response, req)
+ record := p.history.GetRecords()[0]
+ require.Equal(t, tc.outcome, record.Outcome)
+ require.Equal(t, tc.status, record.ResponseStatus)
+ require.Equal(t, "upstream", record.ResponseSource)
+ if tc.outcome == "complete" {
+ require.Contains(t, response.Body.String(), "fixture-response", "redaction must not change forwarded response")
+ }
+ admin := httptest.NewRecorder()
+ p.handleRequestHistory(admin, httptest.NewRequest("GET", "/requests", nil))
+ require.NotContains(t, admin.Body.String(), "fixture-")
+ require.Equal(t, "no-store", admin.Header().Get("Cache-Control"))
+ })
+ }
+}
+
+func TestInspectionOutcomes(t *testing.T) {
+ upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/broken" {
+ w.Header().Set("Content-Length", "999")
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _, err := io.WriteString(w, `{"ok":true}`)
+ require.NoError(t, err)
+ }))
+ defer upstream.Close()
+ p, client, _ := inspectionClient(t, upstream)
+ resp, err := client.Get(upstream.URL + "/broken")
+ if err == nil {
+ _, _ = io.ReadAll(resp.Body)
+ require.NoError(t, resp.Body.Close())
+ }
+ record := waitRecords(t, p, 1)[0]
+ require.Equal(t, "interrupted", record.Outcome)
+ require.Equal(t, 200, record.ResponseStatus)
+ require.Equal(t, "upstream", record.ResponseSource)
+ req, err := http.NewRequest("GET", upstream.URL+"/upgrade", nil)
+ require.NoError(t, err)
+ req.Header.Set("Upgrade", "websocket")
+ req.Header.Set("Connection", "Upgrade")
+ resp, err = client.Do(req)
+ require.NoError(t, err)
+ require.NoError(t, resp.Body.Close())
+ record = waitRecords(t, p, 2)[0]
+ require.Equal(t, "blocked_by_inspection", record.Outcome)
+ require.Equal(t, "proxy", record.ResponseSource)
+ require.Equal(t, 501, record.ResponseStatus)
+}
+
+func TestHeaderHistogramExcludesFailedAttemptsAndTunnels(t *testing.T) {
+ h := NewRequestHistory(3)
+ start := time.Now()
+ for _, tc := range []struct{ outcome, source, method string }{{"complete", "upstream", "GET"}, {"upstream_error", "proxy", "GET"}, {"tunnel_only", "proxy", "CONNECT"}} {
+ h.AddRecord(RequestRecord{Method: tc.method, Outcome: tc.outcome, ResponseSource: tc.source, ResponseStatus: 200, ProxyStartTime: start, ProxyEndTime: start.Add(time.Second), UpstreamStartTime: start, UpstreamEndTime: start.Add(50 * time.Millisecond)})
+ }
+ metrics := h.Metrics()
+ require.Contains(t, metrics, `netkit_response_headers_seconds_count{method="GET",outcome="complete",http_status_class="2xx"} 1`)
+ require.NotContains(t, metrics, `netkit_response_headers_seconds_count{method="CONNECT"`)
+ require.NotContains(t, metrics, `netkit_response_headers_seconds_count{method="GET",outcome="upstream_error"`)
+}
+
+func TestCanceledHTTPExchangeHasBoundedOutcome(t *testing.T) {
+ p := New(&Config{})
+ req := httptest.NewRequest("GET", "http://127.0.0.1:1/?token=fixture-secret", nil)
+ ctx, cancel := context.WithCancel(req.Context())
+ cancel()
+ response := httptest.NewRecorder()
+ p.ServeHTTP(response, req.WithContext(ctx))
+ record := p.history.GetRecords()[0]
+ require.Equal(t, "client_canceled", record.Outcome)
+ require.Equal(t, "proxy", record.ResponseSource)
+ require.Equal(t, 502, record.ResponseStatus)
+ require.NotContains(t, record.URL, "fixture-secret")
+ require.NotContains(t, record.Error, "fixture-secret")
+}
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
index 41fe516..6b65449 100644
--- a/test/e2e/e2e_test.go
+++ b/test/e2e/e2e_test.go
@@ -29,6 +29,32 @@ const (
timeoutDuration = 10 * time.Second
)
+var binaryPath string
+
+// Build current sources once per suite rather than reusing a stale bin/netkit.
+func TestMain(m *testing.M) {
+ buildDir, err := os.MkdirTemp("", "netkit-e2e-")
+ if err != nil {
+ fmt.Fprintln(os.Stderr, err)
+ os.Exit(1)
+ }
+ binaryPath = filepath.Join(buildDir, "netkit")
+ buildCmd := exec.Command("go", "build", "-race", "-o", binaryPath, "../../cmd/netkit")
+ buildCmd.Stdout = os.Stdout
+ buildCmd.Stderr = os.Stderr
+ code := 1
+ if err := buildCmd.Run(); err != nil {
+ fmt.Fprintln(os.Stderr, "Failed to build netkit:", err)
+ } else {
+ code = m.Run()
+ }
+ if err := os.RemoveAll(buildDir); err != nil {
+ fmt.Fprintln(os.Stderr, "Failed to remove test binary:", err)
+ code = 1
+ }
+ os.Exit(code)
+}
+
type testServer struct {
server *http.Server
}
@@ -157,24 +183,6 @@ type netkitCmd struct {
// startProxyServer starts the netkit proxy server with the given arguments
func startProxyServer(t *testing.T, args ...string) *netkitCmd {
- // Build binary path - we assume the binary is in the bin directory
- binaryPath := filepath.Join("..", "..", "bin", "netkit")
-
- // Check if binary exists
- if _, err := os.Stat(binaryPath); os.IsNotExist(err) {
- // Try to build it
- t.Log("Netkit binary not found, building it")
- // Ensure bin directory exists
- binDir := filepath.Join("..", "..", "bin")
- if err := os.MkdirAll(binDir, 0755); err != nil {
- t.Fatalf("Failed to create bin directory: %v", err)
- }
- buildCmd := exec.Command("go", "build", "-o", binaryPath, "../../cmd/netkit")
- if err := buildCmd.Run(); err != nil {
- t.Fatalf("Failed to build netkit binary: %v", err)
- }
- }
-
// Default arguments for the serve command
serveArgs := []string{"serve", fmt.Sprintf("--port=%d", proxyPort), "--log-level=debug"}
@@ -597,7 +605,7 @@ func TestRequestHistoryEndpoints(t *testing.T) {
mostRecent := records[0].(map[string]interface{})
assert.Equal(t, "POST", mostRecent["method"])
assert.Contains(t, mostRecent["url"], "/echo")
- assert.Equal(t, `{"test": "data"}`, mostRecent["request_body"])
+ assert.JSONEq(t, `{"test": "data"}`, mostRecent["request_body"].(string))
assert.Equal(t, float64(200), mostRecent["response_status"])
assert.True(t, mostRecent["success"].(bool))