mirror of
https://github.com/BaaRai/ARTEX.git
synced 2026-11-13 11:40:53 +08:00
Traffic ingest: capture operator browsing via a browser extension
Closes #33. The MITM proxy only sees the agent's own actions, so anything behind a human gate (signup captcha, email OTP, 2FA, SSO) left the authenticated surface dark. The operator's browser becomes a capture source instead: - traffic: exported RecordExternal storing an externally captured exchange through the same spill/blob/FTS path as proxy traffic (insert logic shared with record()); rejects unparsable URLs and empty methods - server: POST /api/traffic/ingest inside the authenticated API group, batch capped at 50 entries, bodies truncated at 256 KiB, per-entry ingested/failed counts - extension/: Manifest V3 Chrome/Edge extension driving the chrome.debugger Network domain on the chosen tab, batching text exchanges to the ingest endpoint with the operator's URL + JWT in chrome.storage.local; binary bodies skipped; flush on batch/stop - tests for both layers, README section
This commit is contained in:
BaaRai
@@ -54,6 +54,15 @@ The global "Approval Records", the in-task "Interception Approvals", and the app
|
||||
[AegisHook approval details component](https://github.com/RuoJi6/AegisHook/blob/main/web/src/components/CallDetail.vue), reusing ARTEX's components and theme:
|
||||
|
||||
|
||||
## Traffic capture browser extension
|
||||
|
||||
The bundled browser extension (`extension/`) turns your own browsing into agent-visible evidence: log into an authenticated target once (captchas, email OTP and 2FA included), and the captured requests/responses land in ARTEX's traffic store like proxy-recorded traffic, ready to search and bind to findings:
|
||||
|
||||
- Load the unpacked extension from `extension/` (Chrome/Edge, Developer Mode → Load unpacked);
|
||||
- In the popup, set your ARTEX URL and token, then press **Start capture** on the tab you are browsing;
|
||||
- Use the site normally; text requests/responses are batched to `POST /api/traffic/ingest` (authenticated). Binary bodies (images, fonts) are skipped, bodies are truncated at 256 KiB, batches are capped at 50 entries;
|
||||
- Press **Stop** when done. Everything stays in your local traffic store.
|
||||
|
||||
## Asset sync (ScopeSentry)
|
||||
|
||||
Sync asset data directly from [ScopeSentry](https://github.com/Autumn-27/ScopeSentry), avoiding duplicate collection:
|
||||
|
||||
@@ -0,0 +1,234 @@
|
||||
// ARTEX Traffic Capture — background service worker.
|
||||
//
|
||||
// Captures the network activity of one tab through the chrome.debugger API
|
||||
// (Network domain) and pushes text request/response pairs to the ARTEX
|
||||
// ingest endpoint in batches. State lives in chrome.storage.session so the
|
||||
// popup and the (short-lived) service worker share one view; the queue is
|
||||
// deliberately in-memory — a browser restart drops unsent entries, which is
|
||||
// the acceptable failure mode for a manual capture session.
|
||||
|
||||
const FLUSH_AT = 10; // entries per batch
|
||||
const MAX_BODY = 256 * 1024; // matches the server-side ingest cap
|
||||
const SKIP_MIME = ["image/", "font/", "audio/", "video/", "application/octet-stream"];
|
||||
|
||||
let session = {
|
||||
capturing: false,
|
||||
tabId: null,
|
||||
queue: [], // pending ExternalExchange objects
|
||||
requests: new Map(), // debugger requestId -> partial request info
|
||||
ingested: 0,
|
||||
failed: 0,
|
||||
lastError: "",
|
||||
};
|
||||
|
||||
async function saveState() {
|
||||
await chrome.storage.session.set({
|
||||
capturing: session.capturing,
|
||||
tabId: session.tabId,
|
||||
ingested: session.ingested,
|
||||
failed: session.failed,
|
||||
lastError: session.lastError,
|
||||
queued: session.queue.length,
|
||||
});
|
||||
}
|
||||
|
||||
async function getConfig() {
|
||||
const { serverUrl = "", token = "" } = await chrome.storage.local.get(["serverUrl", "token"]);
|
||||
return { serverUrl: serverUrl.replace(/\/+$/, ""), token };
|
||||
}
|
||||
|
||||
async function getActiveTabId() {
|
||||
const [tab] = await chrome.tabs.query({ active: true, currentWindow: true });
|
||||
return tab ? tab.id : null;
|
||||
}
|
||||
|
||||
async function start() {
|
||||
if (session.capturing) return { ok: true };
|
||||
const { serverUrl, token } = await getConfig();
|
||||
if (!serverUrl || !token) {
|
||||
return { ok: false, error: "Set the ARTEX URL and token first." };
|
||||
}
|
||||
const tabId = await getActiveTabId();
|
||||
if (tabId == null) return { ok: false, error: "No active tab." };
|
||||
try {
|
||||
await chrome.debugger.attach({ tabId }, "1.3");
|
||||
} catch (e) {
|
||||
return { ok: false, error: `attach failed: ${e.message}` };
|
||||
}
|
||||
await chrome.debugger.sendCommand({ tabId }, "Network.enable", {});
|
||||
session = { ...session, capturing: true, tabId, queue: [], requests: new Map(), ingested: 0, failed: 0, lastError: "" };
|
||||
chrome.action.setBadgeText({ text: "REC" });
|
||||
chrome.action.setBadgeBackgroundColor({ color: "#c2410c" });
|
||||
await saveState();
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
async function stop() {
|
||||
if (session.capturing && session.tabId != null) {
|
||||
await flush();
|
||||
try {
|
||||
await chrome.debugger.sendCommand({ tabId: session.tabId }, "Network.disable", {});
|
||||
await chrome.debugger.detach({ tabId: session.tabId });
|
||||
} catch (_) {
|
||||
// tab may already be gone; nothing to do
|
||||
}
|
||||
}
|
||||
session.capturing = false;
|
||||
session.tabId = null;
|
||||
chrome.action.setBadgeText({ text: "" });
|
||||
await saveState();
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
function skipBody(mimeType) {
|
||||
const mt = (mimeType || "").toLowerCase();
|
||||
return SKIP_MIME.some((p) => mt.startsWith(p));
|
||||
}
|
||||
|
||||
function truncate(s) {
|
||||
if (s.length <= MAX_BODY) return s;
|
||||
return s.slice(0, MAX_BODY) + "\n... [truncated by extension]";
|
||||
}
|
||||
|
||||
async function queueExchange(info, response) {
|
||||
const respHeaders = {};
|
||||
for (const h of response.headers || []) respHeaders[h.name.toLowerCase()] = h.value;
|
||||
const mime = respHeaders["content-type"] || "";
|
||||
|
||||
let reqBody = info.postData || "";
|
||||
let respBody = "";
|
||||
if (!skipBody(mime)) {
|
||||
try {
|
||||
const body = await chrome.debugger.sendCommand(
|
||||
{ tabId: session.tabId },
|
||||
"Network.getResponseBody",
|
||||
{ requestId: info.requestId }
|
||||
);
|
||||
if (body && !body.base64Encoded) {
|
||||
reqBody = truncate(reqBody);
|
||||
respBody = truncate(body.body || "");
|
||||
}
|
||||
} catch (_) {
|
||||
// body already discarded by the browser; capture the pair without bodies
|
||||
}
|
||||
}
|
||||
|
||||
session.queue.push({
|
||||
method: info.method,
|
||||
url: info.url,
|
||||
status: response.status || 0,
|
||||
requestHeaders: info.headers || {},
|
||||
requestBody: reqBody,
|
||||
responseHeaders: respHeaders,
|
||||
responseBody: respBody,
|
||||
});
|
||||
if (session.queue.length >= FLUSH_AT) await flush();
|
||||
}
|
||||
|
||||
async function flush() {
|
||||
if (session.queue.length === 0) return;
|
||||
const { serverUrl, token } = await getConfig();
|
||||
if (!serverUrl || !token) {
|
||||
session.lastError = "missing server URL or token";
|
||||
await saveState();
|
||||
return;
|
||||
}
|
||||
const batch = session.queue.splice(0, FLUSH_AT);
|
||||
try {
|
||||
const res = await fetch(`${serverUrl}/api/traffic/ingest`, {
|
||||
method: "POST",
|
||||
headers: { "Authorization": `Bearer ${token}`, "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ exchanges: batch }),
|
||||
});
|
||||
if (!res.ok) throw new Error(`HTTP ${res.status}: ${(await res.text()).slice(0, 200)}`);
|
||||
const out = await res.json();
|
||||
session.ingested += out.ingested || 0;
|
||||
session.failed += out.failed || 0;
|
||||
session.lastError = out.failed ? `${out.failed} entries rejected by the server` : "";
|
||||
} catch (e) {
|
||||
// put the batch back (oldest first) but cap the queue so a dead server
|
||||
// cannot grow memory unbounded
|
||||
session.queue = batch.concat(session.queue).slice(0, 200);
|
||||
session.lastError = e.message;
|
||||
}
|
||||
await saveState();
|
||||
}
|
||||
|
||||
chrome.debugger.onEvent.addListener((source, method, params) => {
|
||||
if (!session.capturing || !source.tabId || source.tabId !== session.tabId) return;
|
||||
if (method === "Network.requestWillBeSent") {
|
||||
session.requests.set(params.requestId, {
|
||||
requestId: params.requestId,
|
||||
method: params.request.method,
|
||||
url: params.request.url,
|
||||
headers: params.request.headers,
|
||||
postData: params.request.postData || "",
|
||||
});
|
||||
} else if (method === "Network.responseReceived") {
|
||||
const info = session.requests.get(params.requestId);
|
||||
if (info) info.response = params.response;
|
||||
} else if (method === "Network.loadingFinished") {
|
||||
const info = session.requests.get(params.requestId);
|
||||
session.requests.delete(params.requestId);
|
||||
if (!info || !info.response) return;
|
||||
queueExchange(info, info.response).catch((e) => {
|
||||
session.lastError = e.message;
|
||||
saveState();
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
chrome.debugger.onDetach.addListener((source) => {
|
||||
if (session.capturing && source.tabId === session.tabId) {
|
||||
// DevTools opened, or the tab navigated somewhere undebaggable: stop
|
||||
// cleanly instead of silently dropping events.
|
||||
session.capturing = false;
|
||||
session.tabId = null;
|
||||
session.lastError = "debugger detached (DevTools opened?)";
|
||||
chrome.action.setBadgeText({ text: "" });
|
||||
saveState();
|
||||
}
|
||||
});
|
||||
|
||||
chrome.runtime.onMessage.addListener((msg, _sender, sendResponse) => {
|
||||
(async () => {
|
||||
switch (msg?.type) {
|
||||
case "getState": {
|
||||
sendResponse({
|
||||
capturing: session.capturing,
|
||||
tabId: session.tabId,
|
||||
ingested: session.ingested,
|
||||
failed: session.failed,
|
||||
queued: session.queue.length,
|
||||
lastError: session.lastError,
|
||||
configured: !!(await getConfig()).serverUrl,
|
||||
});
|
||||
return;
|
||||
}
|
||||
case "start":
|
||||
sendResponse(await start());
|
||||
return;
|
||||
case "stop":
|
||||
sendResponse(await stop());
|
||||
return;
|
||||
case "flushNow":
|
||||
await flush();
|
||||
sendResponse(await stateSnapshot());
|
||||
return;
|
||||
default:
|
||||
sendResponse({ ok: false, error: "unknown message" });
|
||||
}
|
||||
})();
|
||||
return true; // async sendResponse
|
||||
});
|
||||
|
||||
async function stateSnapshot() {
|
||||
return {
|
||||
capturing: session.capturing,
|
||||
tabId: session.tabId,
|
||||
ingested: session.ingested,
|
||||
failed: session.failed,
|
||||
queued: session.queue.length,
|
||||
lastError: session.lastError,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"manifest_version": 3,
|
||||
"name": "ARTEX Traffic Capture",
|
||||
"version": "1.0.0",
|
||||
"description": "Capture the current tab's network traffic into your local ARTEX instance, so operator-driven authenticated browsing becomes agent-visible evidence.",
|
||||
"permissions": ["debugger", "storage", "tabs"],
|
||||
"host_permissions": ["<all_urls>"],
|
||||
"background": { "service_worker": "background.js" },
|
||||
"action": { "default_popup": "popup.html", "default_title": "ARTEX Traffic Capture" }
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<meta charset="utf-8" />
|
||||
<style>
|
||||
body { font: 13px system-ui, sans-serif; width: 320px; margin: 0; padding: 12px; }
|
||||
h1 { font-size: 14px; margin: 0 0 10px; }
|
||||
label { display: block; margin: 8px 0 2px; color: #444; }
|
||||
input { width: 100%; box-sizing: border-box; padding: 5px; }
|
||||
button { margin-top: 10px; padding: 6px 12px; cursor: pointer; }
|
||||
#start { background: #1d4ed8; color: #fff; border: 0; border-radius: 4px; }
|
||||
#stop { background: #eee; border: 1px solid #ccc; border-radius: 4px; }
|
||||
#status { margin-top: 10px; line-height: 1.5; color: #333; }
|
||||
#error { color: #b91c1c; white-space: pre-wrap; }
|
||||
.muted { color: #777; font-size: 11px; margin-top: 10px; line-height: 1.4; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<h1>ARTEX Traffic Capture</h1>
|
||||
|
||||
<label for="serverUrl">ARTEX URL</label>
|
||||
<input id="serverUrl" placeholder="http://localhost:8787" />
|
||||
<label for="token">Token</label>
|
||||
<input id="token" type="password" placeholder="ARTEX JWT (localStorage artex_token)" />
|
||||
|
||||
<button id="save">Save settings</button>
|
||||
<button id="start">Start capture</button>
|
||||
<button id="stop">Stop</button>
|
||||
<button id="flush">Flush now</button>
|
||||
|
||||
<div id="status"></div>
|
||||
<div id="error"></div>
|
||||
|
||||
<p class="muted">
|
||||
Start on the tab you are browsing, log in and use the app normally;
|
||||
captured text requests/responses land in your local ARTEX traffic store
|
||||
(Traffic page) as evidence. Stop when done. Binary bodies (images, fonts)
|
||||
are skipped.
|
||||
</p>
|
||||
|
||||
<script src="popup.js"></script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,46 @@
|
||||
const $ = (id) => document.getElementById(id);
|
||||
|
||||
async function refreshState() {
|
||||
const st = await chrome.runtime.sendMessage({ type: "getState" });
|
||||
const lines = [];
|
||||
lines.push(st.capturing
|
||||
? `● Recording tab ${st.tabId}`
|
||||
: "○ Not recording");
|
||||
lines.push(`ingested: ${st.ingested} · queued: ${st.queued} · failed: ${st.failed}`);
|
||||
if (st.lastError) lines.push(st.lastError);
|
||||
$("status").textContent = lines.join("\n");
|
||||
$("error").textContent = st.capturing ? "" : (st.lastError || "");
|
||||
}
|
||||
|
||||
$("save").addEventListener("click", async () => {
|
||||
await chrome.storage.local.set({
|
||||
serverUrl: $("serverUrl").value.trim(),
|
||||
token: $("token").value.trim(),
|
||||
});
|
||||
$("save").textContent = "Saved ✓";
|
||||
setTimeout(() => ($("save").textContent = "Save settings"), 1200);
|
||||
});
|
||||
|
||||
$("start").addEventListener("click", async () => {
|
||||
const out = await chrome.runtime.sendMessage({ type: "start" });
|
||||
if (out && out.ok === false) $("error").textContent = out.error || "failed to start";
|
||||
refreshState();
|
||||
});
|
||||
|
||||
$("stop").addEventListener("click", async () => {
|
||||
await chrome.runtime.sendMessage({ type: "stop" });
|
||||
refreshState();
|
||||
});
|
||||
|
||||
$("flush").addEventListener("click", async () => {
|
||||
await chrome.runtime.sendMessage({ type: "flushNow" });
|
||||
refreshState();
|
||||
});
|
||||
|
||||
(async () => {
|
||||
const { serverUrl = "", token = "" } = await chrome.storage.local.get(["serverUrl", "token"]);
|
||||
$("serverUrl").value = serverUrl;
|
||||
$("token").value = token;
|
||||
refreshState();
|
||||
setInterval(refreshState, 1500);
|
||||
})();
|
||||
@@ -793,6 +793,7 @@ func (s *Server) Handler() http.Handler {
|
||||
mux.HandleFunc("DELETE /api/traffic/all", s.deleteAllTraffic)
|
||||
mux.HandleFunc("GET /api/traffic/exchange", s.getTrafficExchange)
|
||||
mux.HandleFunc("GET /api/traffic/blob", s.getTrafficBlob)
|
||||
mux.HandleFunc("POST /api/traffic/ingest", s.ingestTraffic)
|
||||
mux.HandleFunc("GET /api/commands", s.pgListCommands)
|
||||
mux.HandleFunc("GET /api/commands/stats", s.pgToolStats) // aggregate call counts per tool
|
||||
mux.HandleFunc("GET /api/llm/records", s.pgListLLMRecords)
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/Autumn-27/artex/traffic"
|
||||
)
|
||||
|
||||
// Ingest caps: a browser tab can produce a burst of entries, but the ingest
|
||||
// path is meant for deliberate operator captures, not bulk sync. Bodies are
|
||||
// truncated server-side as a second line of defense behind the extension's
|
||||
// own caps.
|
||||
const (
|
||||
maxIngestBatch = 50
|
||||
maxIngestBodySize = 256 << 10 // 256 KiB per body
|
||||
)
|
||||
|
||||
// ingestTraffic accepts exchanges captured outside the MITM proxy (currently
|
||||
// the browser extension) into the global traffic store, so operator-driven
|
||||
// authenticated browsing becomes evidence the same way proxy traffic is.
|
||||
func (s *Server) ingestTraffic(w http.ResponseWriter, r *http.Request) {
|
||||
tr := s.m.Traffic()
|
||||
if tr == nil {
|
||||
writeErr(w, 503, "traffic store is not available")
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
Exchanges []traffic.ExternalExchange `json:"exchanges"`
|
||||
}
|
||||
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 32<<20)).Decode(&req); err != nil {
|
||||
writeErr(w, 400, "bad json: "+err.Error())
|
||||
return
|
||||
}
|
||||
if len(req.Exchanges) == 0 {
|
||||
writeErr(w, 400, "empty exchanges")
|
||||
return
|
||||
}
|
||||
if len(req.Exchanges) > maxIngestBatch {
|
||||
writeErr(w, 400, "too many exchanges in one batch")
|
||||
return
|
||||
}
|
||||
|
||||
ingested, failed := 0, 0
|
||||
for _, e := range req.Exchanges {
|
||||
e.RequestBody = truncateBody(e.RequestBody)
|
||||
e.ResponseBody = truncateBody(e.ResponseBody)
|
||||
if err := tr.RecordExternal(e); err != nil {
|
||||
failed++
|
||||
continue
|
||||
}
|
||||
ingested++
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{"ingested": ingested, "failed": failed})
|
||||
}
|
||||
|
||||
func truncateBody(s string) string {
|
||||
if len(s) <= maxIngestBodySize {
|
||||
return s
|
||||
}
|
||||
return s[:maxIngestBodySize] + "\n... [truncated by ingest]"
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/Autumn-27/artex/traffic"
|
||||
)
|
||||
|
||||
// ingestServer wires a Server with a live traffic store, like
|
||||
// trafficEvidenceServer but without the finding plumbing.
|
||||
func ingestServer(t *testing.T) (*Server, func(method, path, body string) *httptest.ResponseRecorder) {
|
||||
t.Helper()
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { m.Close() })
|
||||
m.traffic, err = traffic.Open(filepath.Join(m.dir, "traffic"), ":0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
s := New(ctx, m, t.TempDir(), t.TempDir(), t.TempDir())
|
||||
s.archiveWG.Wait()
|
||||
<-s.side.done
|
||||
token, err := signJWT(s.jwtKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return s, func(method, path, body string) *httptest.ResponseRecorder {
|
||||
r := httptest.NewRequest(method, path, strings.NewReader(body))
|
||||
r.Header.Set("Authorization", "Bearer "+token)
|
||||
r.Header.Set("Content-Type", "application/json")
|
||||
w := httptest.NewRecorder()
|
||||
s.Handler().ServeHTTP(w, r)
|
||||
return w
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrafficIngestFlow(t *testing.T) {
|
||||
s, request := ingestServer(t)
|
||||
|
||||
body := fmt.Sprintf(`{"exchanges":[
|
||||
{"method":"POST","url":"https://api.example.com/v1/login","status":200,
|
||||
"requestHeaders":{"content-type":"application/json"},"requestBody":"{\"u\":\"u1\"}",
|
||||
"responseHeaders":{"content-type":"application/json"},"responseBody":"{\"token\":\"t\"}"},
|
||||
{"method":"GET","url":"https://api.example.com/v1/profile","status":200,
|
||||
"requestHeaders":{},"requestBody":"",
|
||||
"responseHeaders":{},"responseBody":"{\"name\":\"n\"}"}
|
||||
]}`)
|
||||
w := request("POST", "/api/traffic/ingest", body)
|
||||
if w.Code != 200 {
|
||||
t.Fatalf("ingest status = %d, body = %s", w.Code, w.Body)
|
||||
}
|
||||
if !strings.Contains(w.Body.String(), `"ingested":2`) || !strings.Contains(w.Body.String(), `"failed":0`) {
|
||||
t.Fatalf("unexpected counts: %s", w.Body)
|
||||
}
|
||||
|
||||
// Both entries are visible on the normal traffic API, like proxy traffic.
|
||||
w = request("GET", "/api/traffic", "")
|
||||
if w.Code != 200 || !strings.Contains(w.Body.String(), "api.example.com") {
|
||||
t.Fatalf("traffic page after ingest: %d %s", w.Code, w.Body)
|
||||
}
|
||||
|
||||
// One bad URL among good ones: the batch reports per-entry failure.
|
||||
mixed := `{"exchanges":[
|
||||
{"method":"GET","url":"https://good.example.com/ok","status":200,"responseBody":"ok"},
|
||||
{"method":"GET","url":"::::","status":200,"responseBody":"bad"}
|
||||
]}`
|
||||
w = request("POST", "/api/traffic/ingest", mixed)
|
||||
if w.Code != 200 || !strings.Contains(w.Body.String(), `"ingested":1`) || !strings.Contains(w.Body.String(), `"failed":1`) {
|
||||
t.Fatalf("mixed batch: %d %s", w.Code, w.Body)
|
||||
}
|
||||
|
||||
// Guards: empty batch, oversized batch, oversized body truncation.
|
||||
if w = request("POST", "/api/traffic/ingest", `{"exchanges":[]}`); w.Code != 400 {
|
||||
t.Fatalf("empty batch status = %d", w.Code)
|
||||
}
|
||||
big := `{"exchanges":[` + strings.Repeat(`{"method":"GET","url":"https://x.example.com/"},`, maxIngestBatch) + `{}]}`
|
||||
if w = request("POST", "/api/traffic/ingest", big); w.Code != 400 {
|
||||
t.Fatalf("oversized batch status = %d", w.Code)
|
||||
}
|
||||
trunc := `{"exchanges":[{"method":"GET","url":"https://big.example.com/","status":200,"responseBody":"` + strings.Repeat("x", maxIngestBodySize+1024) + `"}]}`
|
||||
w = request("POST", "/api/traffic/ingest", trunc)
|
||||
if w.Code != 200 || !strings.Contains(w.Body.String(), `"ingested":1`) {
|
||||
t.Fatalf("truncating batch: %d %s", w.Code, w.Body)
|
||||
}
|
||||
if n, _ := s.m.Traffic().Count(); n != 4 { // 2 + 1 + 1
|
||||
t.Fatalf("final count = %d, want 4", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrafficIngestRequiresAuth(t *testing.T) {
|
||||
s, _ := ingestServer(t)
|
||||
r := httptest.NewRequest("POST", "/api/traffic/ingest", strings.NewReader(`{"exchanges":[]}`))
|
||||
w := httptest.NewRecorder()
|
||||
s.Handler().ServeHTTP(w, r)
|
||||
if w.Code != 401 {
|
||||
t.Fatalf("unauthenticated ingest status = %d, want 401", w.Code)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package traffic
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Autumn-27/artex/db"
|
||||
)
|
||||
|
||||
// ExternalExchange is one captured request/response pair handed in from
|
||||
// outside the MITM proxy (currently the browser-extension ingest endpoint).
|
||||
// Header maps are lowercased-name oriented on the extension side but any
|
||||
// casing is accepted; bodies are raw text (binary capture is skipped by the
|
||||
// extension, not handled here).
|
||||
type ExternalExchange struct {
|
||||
Method string `json:"method"`
|
||||
URL string `json:"url"`
|
||||
Status int `json:"status"`
|
||||
RequestHeaders map[string]string `json:"requestHeaders"`
|
||||
RequestBody string `json:"requestBody"`
|
||||
ResponseHeaders map[string]string `json:"responseHeaders"`
|
||||
ResponseBody string `json:"responseBody"`
|
||||
}
|
||||
|
||||
// RecordExternal stores an externally captured exchange through the same
|
||||
// spill/blob/index path as proxy traffic, so ingested entries are
|
||||
// indistinguishable from proxy-captured ones on the Traffic page and in
|
||||
// full-text search.
|
||||
func (t *Traffic) RecordExternal(e ExternalExchange) error {
|
||||
u, err := url.Parse(e.URL)
|
||||
if err != nil || u.Host == "" {
|
||||
return fmt.Errorf("invalid url %q", e.URL)
|
||||
}
|
||||
method := strings.ToUpper(strings.TrimSpace(e.Method))
|
||||
if method == "" {
|
||||
return fmt.Errorf("empty method for %q", e.URL)
|
||||
}
|
||||
status := e.Status
|
||||
if status == 0 {
|
||||
status = 200
|
||||
}
|
||||
|
||||
host := u.Hostname()
|
||||
reqHead := fmt.Sprintf("%s %s %s\n%s", method, u.RequestURI(), "HTTP/1.1", headerLines(toHeaderMap(e.RequestHeaders)))
|
||||
respHead := fmt.Sprintf("HTTP %d\n%s", status, headerLines(toHeaderMap(e.ResponseHeaders)))
|
||||
respCT := e.ResponseHeaders["content-type"]
|
||||
|
||||
t.wmu.Lock()
|
||||
defer t.wmu.Unlock()
|
||||
|
||||
// Bodies are spilled before the transaction opens, matching record().
|
||||
reqB := t.spill([]byte(e.RequestBody), e.RequestHeaders["content-type"])
|
||||
respB := t.spill([]byte(e.ResponseBody), respCT)
|
||||
|
||||
n := t.seq.Add(1)
|
||||
now := time.Now()
|
||||
id := fmt.Sprintf("%d-%04d", now.Unix(), n%10000)
|
||||
|
||||
if err := t.insertExchange(id, now.Unix(), host, method, db.TemplatePath(u.EscapedPath()), u.String(), status, respCT, len(e.RequestBody), len(e.ResponseBody), reqHead, reqB, respHead, respB); err != nil {
|
||||
log.Printf("[traffic] failed to ingest %s: %v", u.String(), err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// insertExchange writes the index row, the body row, blob references and the
|
||||
// FTS entry inside one transaction. record() and RecordExternal share it so
|
||||
// both capture paths stay byte-identical in storage layout.
|
||||
func (t *Traffic) insertExchange(id string, ts int64, host, method, urlTemplate, rawURL string, status int, contentType string, reqLen, respLen int, reqHead string, reqB storedBody, respHead string, respB storedBody) error {
|
||||
tx, err := t.db.Begin()
|
||||
if err != nil {
|
||||
return fmt.Errorf("beginning transaction: %w", err)
|
||||
}
|
||||
defer tx.Rollback() //nolint:errcheck // no-op once committed
|
||||
|
||||
res, err := tx.Exec(`INSERT OR REPLACE INTO exchanges(id,ts,host,method,url_template,url,status,content_type,req_len,resp_len,path)
|
||||
VALUES(?,?,?,?,?,?,?,?,?,?,'')`,
|
||||
id, ts, host, method, urlTemplate, rawURL, status, contentType, reqLen, respLen)
|
||||
if err != nil {
|
||||
return fmt.Errorf("writing index: %w", err)
|
||||
}
|
||||
rowid, err := res.LastInsertId()
|
||||
if err != nil {
|
||||
return fmt.Errorf("fetching rowid: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(`INSERT OR REPLACE INTO exchange_bodies(id,req_head,req_body,req_blob,resp_head,resp_body,resp_blob)
|
||||
VALUES(?,?,?,?,?,?,?)`,
|
||||
id, reqHead, reqB.inline, nullIfEmpty(reqB.hash), respHead, respB.inline, nullIfEmpty(respB.hash)); err != nil {
|
||||
return fmt.Errorf("writing body: %w", err)
|
||||
}
|
||||
|
||||
for _, h := range []string{reqB.hash, respB.hash} {
|
||||
if h == "" {
|
||||
continue
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT OR IGNORE INTO blob_refs(hash,exchange_id) VALUES(?,?)`, h, id); err != nil {
|
||||
return fmt.Errorf("registering blob reference: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if t.fts {
|
||||
idx := strings.Join([]string{rawURL, reqHead, reqB.index, respHead, respB.index}, "\n")
|
||||
if _, err := tx.Exec(`INSERT INTO ex_fts(rowid,content) VALUES(?,?)`, rowid, idx); err != nil {
|
||||
return fmt.Errorf("writing full-text index: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("committing: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func toHeaderMap(flat map[string]string) map[string][]string {
|
||||
m := make(map[string][]string, len(flat))
|
||||
for k, v := range flat {
|
||||
m[k] = []string{v}
|
||||
}
|
||||
return m
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package traffic
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestRecordExternalStoresLikeProxyTraffic(t *testing.T) {
|
||||
tr, _ := openTraffic(t)
|
||||
|
||||
big := strings.Repeat("A", 2<<20) // spills to the blob store
|
||||
err := tr.RecordExternal(ExternalExchange{
|
||||
Method: "POST",
|
||||
URL: "https://api.example.com/v1/login",
|
||||
Status: 200,
|
||||
RequestHeaders: map[string]string{"content-type": "application/json", "authorization": "Bearer xyz"},
|
||||
RequestBody: `{"user":"u1","password":"p1"}`,
|
||||
ResponseHeaders: map[string]string{"content-type": "application/json"},
|
||||
ResponseBody: `{"token":"tok"}`,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("RecordExternal small: %v", err)
|
||||
}
|
||||
if err := tr.RecordExternal(ExternalExchange{
|
||||
Method: "GET",
|
||||
URL: "https://api.example.com/v1/items?page=1",
|
||||
Status: 404,
|
||||
ResponseHeaders: map[string]string{"content-type": "text/html"},
|
||||
ResponseBody: big,
|
||||
}); err != nil {
|
||||
t.Fatalf("RecordExternal spilled: %v", err)
|
||||
}
|
||||
|
||||
if n, _ := tr.Count(); n != 2 {
|
||||
t.Fatalf("count = %d, want 2", n)
|
||||
}
|
||||
rows, total, err := tr.Page(PageQuery{Query: "login", RespMin: -1, RespMax: -1}, 0, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if total != 1 || len(rows) != 1 {
|
||||
t.Fatalf("filtered page = %d rows (total %d), want 1", len(rows), total)
|
||||
}
|
||||
req, resp, err := tr.Get(rows[0].ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(req, "POST /v1/login") || !strings.Contains(req, "Bearer xyz") || !strings.Contains(req, `"user":"u1"`) {
|
||||
t.Fatalf("request side incomplete: %q", req)
|
||||
}
|
||||
if !strings.Contains(resp, `{"token":"tok"}`) {
|
||||
t.Fatalf("response side incomplete: %q", resp)
|
||||
}
|
||||
|
||||
// The spilled entry must be FTS-searchable by its body content too.
|
||||
rows, total, err = tr.Page(PageQuery{Body: strings.Repeat("A", 64), RespMin: -1, RespMax: -1}, 0, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if total != 1 || len(rows) != 1 {
|
||||
t.Fatalf("spilled-body FTS search = %d rows (total %d), want 1", len(rows), total)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordExternalRejectsBadInput(t *testing.T) {
|
||||
tr, _ := openTraffic(t)
|
||||
cases := []ExternalExchange{
|
||||
{Method: "GET", URL: "not a url"},
|
||||
{Method: "GET", URL: ""},
|
||||
{Method: "", URL: "https://api.example.com/"},
|
||||
}
|
||||
for i, c := range cases {
|
||||
if err := tr.RecordExternal(c); err == nil {
|
||||
t.Fatalf("case %d: expected an error for %+v", i, c)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user