mirror of
https://github.com/littledivy/mimic.git
synced 2026-10-26 19:15:10 +00:00
feat: add HAR file import as a capture backend (#5)
The sources/__init__.py docstring promised "cURL/HAR paste" but HAR was never implemented. This adds it, so mimic works with browser devtools exports (Chrome, Firefox, Safari) and desktop proxies (Charles, Proxyman) with no mitmproxy or iPhone setup. - new mimic/sources/har.py parses HAR JSON into mimic's flow dicts, with inline request/response bodies (base64 decoded) for codegen - Session.from_har(path, host) constructor; host defaults to the most-requested one - --har flag on the hosts, learn, and gen commands - tests/test_har.py with a small fixture The HAR-to-flow conversion produces the same dict shape extract.py and codegen.py already consume, so those modules are untouched. Signed-off-by: ZhYGuoL <zguoliau@andrew.cmu.edu>
This commit is contained in:
1 parent
67441bfaba
commit
027bc48698
6 files changed
+298
-9
No files matched your search
@@ -93,6 +93,11 @@ pass `refresh=True`.
|
|||||||
runs it via `uvx`, so there's nothing extra to install.
|
runs it via `uvx`, so there's nothing extra to install.
|
||||||
- **cURL / paste** for anything with a web version. `Copy as cURL` in devtools,
|
- **cURL / paste** for anything with a web version. `Copy as cURL` in devtools,
|
||||||
then `Session.from_curl(text)`. No proxy, no cert.
|
then `Session.from_curl(text)`. No proxy, no cert.
|
||||||
|
- **HAR file** for web apps and anything you can capture in a browser. In Chrome
|
||||||
|
or Firefox devtools, open the Network tab, right-click a request, and choose
|
||||||
|
"Save all as HAR". Then `mimic hosts --har traffic.har` and
|
||||||
|
`mimic gen api.example.com --har traffic.har`, or build a session directly with
|
||||||
|
`Session.from_har("traffic.har", "api.example.com")`. No proxy, no cert.
|
||||||
|
|
||||||
## Limitations
|
## Limitations
|
||||||
|
|
||||||
|
|||||||
+29
-8
@@ -16,6 +16,7 @@ import sys
|
|||||||
|
|
||||||
from . import codegen
|
from . import codegen
|
||||||
from . import unpin
|
from . import unpin
|
||||||
|
from .sources import har
|
||||||
from .sources import mitm
|
from .sources import mitm
|
||||||
|
|
||||||
|
|
||||||
@@ -24,6 +25,20 @@ def _mitm_and_flows():
|
|||||||
return m, m.flows()
|
return m, m.flows()
|
||||||
|
|
||||||
|
|
||||||
|
def _flows_from_args(args):
|
||||||
|
"""Flows from a HAR file when --har is given, else from live mitmweb."""
|
||||||
|
if getattr(args, "har", None):
|
||||||
|
return None, har.load(args.har)
|
||||||
|
return _mitm_and_flows()
|
||||||
|
|
||||||
|
|
||||||
|
def _endpoints_from_args(args, m, flows):
|
||||||
|
"""Endpoints for a host, from HAR (inline bodies) or mitmweb (fetched)."""
|
||||||
|
if getattr(args, "har", None):
|
||||||
|
return har.endpoints(args.har, args.host)
|
||||||
|
return mitm.endpoints(m, flows, args.host)
|
||||||
|
|
||||||
|
|
||||||
def _lan_ip():
|
def _lan_ip():
|
||||||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||||
try:
|
try:
|
||||||
@@ -124,9 +139,11 @@ def cmd_doctor(args):
|
|||||||
|
|
||||||
|
|
||||||
def cmd_hosts(args):
|
def cmd_hosts(args):
|
||||||
_, flows = _mitm_and_flows()
|
_, flows = _flows_from_args(args)
|
||||||
rows = mitm.hosts(flows)
|
rows = mitm.hosts(flows)
|
||||||
if not rows:
|
if not rows:
|
||||||
|
if getattr(args, "har", None):
|
||||||
|
sys.exit("no entries in the HAR file")
|
||||||
sys.exit("no traffic captured yet — run `mimic record` and use the app")
|
sys.exit("no traffic captured yet — run `mimic record` and use the app")
|
||||||
print(f"{'requests':>9} host")
|
print(f"{'requests':>9} host")
|
||||||
for host, n in rows:
|
for host, n in rows:
|
||||||
@@ -135,20 +152,20 @@ def cmd_hosts(args):
|
|||||||
|
|
||||||
|
|
||||||
def cmd_learn(args):
|
def cmd_learn(args):
|
||||||
m, flows = _mitm_and_flows()
|
m, flows = _flows_from_args(args)
|
||||||
eps = mitm.endpoints(m, flows, args.host)
|
eps = _endpoints_from_args(args, m, flows)
|
||||||
if not eps:
|
if not eps:
|
||||||
sys.exit(f"no requests to {args.host} captured")
|
sys.exit(f"no requests to {args.host} found")
|
||||||
print(f"{args.host}: {len(eps)} endpoints\n")
|
print(f"{args.host}: {len(eps)} endpoints\n")
|
||||||
for e in eps:
|
for e in eps:
|
||||||
print(f" {e['method']:5s} {e['path']} -> {e['status']}")
|
print(f" {e['method']:5s} {e['path']} -> {e['status']}")
|
||||||
|
|
||||||
|
|
||||||
def cmd_gen(args):
|
def cmd_gen(args):
|
||||||
m, flows = _mitm_and_flows()
|
m, flows = _flows_from_args(args)
|
||||||
eps = mitm.endpoints(m, flows, args.host)
|
eps = _endpoints_from_args(args, m, flows)
|
||||||
if not eps:
|
if not eps:
|
||||||
sys.exit(f"no requests to {args.host} captured")
|
sys.exit(f"no requests to {args.host} found")
|
||||||
|
|
||||||
if args.prompt_only:
|
if args.prompt_only:
|
||||||
print(codegen.build_prompt(args.host, eps))
|
print(codegen.build_prompt(args.host, eps))
|
||||||
@@ -183,10 +200,13 @@ def main(argv=None):
|
|||||||
|
|
||||||
sub.add_parser("record", help="start the proxy + iPhone setup steps").set_defaults(func=cmd_record)
|
sub.add_parser("record", help="start the proxy + iPhone setup steps").set_defaults(func=cmd_record)
|
||||||
sub.add_parser("doctor", help="check your setup").set_defaults(func=cmd_doctor)
|
sub.add_parser("doctor", help="check your setup").set_defaults(func=cmd_doctor)
|
||||||
sub.add_parser("hosts", help="list captured hosts").set_defaults(func=cmd_hosts)
|
hp = sub.add_parser("hosts", help="list captured hosts")
|
||||||
|
hp.add_argument("--har", metavar="FILE", help="read from a HAR file instead of mitmweb")
|
||||||
|
hp.set_defaults(func=cmd_hosts)
|
||||||
|
|
||||||
lp = sub.add_parser("learn", help="show endpoints for a host")
|
lp = sub.add_parser("learn", help="show endpoints for a host")
|
||||||
lp.add_argument("host")
|
lp.add_argument("host")
|
||||||
|
lp.add_argument("--har", metavar="FILE", help="read from a HAR file instead of mitmweb")
|
||||||
lp.set_defaults(func=cmd_learn)
|
lp.set_defaults(func=cmd_learn)
|
||||||
|
|
||||||
gp = sub.add_parser("gen", help="AI-generate a client for a host")
|
gp = sub.add_parser("gen", help="AI-generate a client for a host")
|
||||||
@@ -196,6 +216,7 @@ def main(argv=None):
|
|||||||
gp.add_argument("--generator", default="claude", choices=["claude", "opencode"],
|
gp.add_argument("--generator", default="claude", choices=["claude", "opencode"],
|
||||||
help="AI generator to use (default: claude)")
|
help="AI generator to use (default: claude)")
|
||||||
gp.add_argument("--prompt-only", action="store_true", help="print the prompt instead of calling the AI generator")
|
gp.add_argument("--prompt-only", action="store_true", help="print the prompt instead of calling the AI generator")
|
||||||
|
gp.add_argument("--har", metavar="FILE", help="read from a HAR file instead of mitmweb")
|
||||||
gp.set_defaults(func=cmd_gen)
|
gp.set_defaults(func=cmd_gen)
|
||||||
|
|
||||||
up = sub.add_parser("unpin", help="defeat cert pinning via Frida so capture works")
|
up = sub.add_parser("unpin", help="defeat cert pinning via Frida so capture works")
|
||||||
|
|||||||
+20
-1
@@ -19,10 +19,11 @@ _IDEMPOTENT_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "PUT", "DELETE"})
|
|||||||
class Session:
|
class Session:
|
||||||
"""Base URL + reusable headers, with helpers that return parsed JSON.
|
"""Base URL + reusable headers, with helpers that return parsed JSON.
|
||||||
|
|
||||||
Construct one of three ways:
|
Construct one of four ways:
|
||||||
Session.from_mitm("prod-api.hingeaws.net") # pull from mitmweb
|
Session.from_mitm("prod-api.hingeaws.net") # pull from mitmweb
|
||||||
Session(base_url=..., headers={...}) # explicit
|
Session(base_url=..., headers={...}) # explicit
|
||||||
Session.from_curl(text) # paste a copied cURL
|
Session.from_curl(text) # paste a copied cURL
|
||||||
|
Session.from_har("session.har") # load a browser HAR export
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, base_url, headers=None, host=None, mitm=None):
|
def __init__(self, base_url, headers=None, host=None, mitm=None):
|
||||||
@@ -50,6 +51,24 @@ class Session:
|
|||||||
base_url, headers = _parse_curl(text)
|
base_url, headers = _parse_curl(text)
|
||||||
return cls(base_url, headers)
|
return cls(base_url, headers)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_har(cls, path, host=None):
|
||||||
|
"""Build a session from a HAR file exported from browser devtools.
|
||||||
|
|
||||||
|
With no host, picks the most-requested host in the file.
|
||||||
|
"""
|
||||||
|
from .sources import har
|
||||||
|
flows = har.load(path)
|
||||||
|
if not host:
|
||||||
|
ranked = har.hosts(path)
|
||||||
|
if not ranked:
|
||||||
|
raise RuntimeError("HAR file has no entries")
|
||||||
|
host = ranked[0][0]
|
||||||
|
headers = extract.session_headers(flows, host)
|
||||||
|
if not headers:
|
||||||
|
raise RuntimeError(f"no authenticated request to {host} in the HAR file")
|
||||||
|
return cls(extract.base_url(flows, host), headers, host=host)
|
||||||
|
|
||||||
# ---- calls --------------------------------------------------------------
|
# ---- calls --------------------------------------------------------------
|
||||||
def request(self, method, path, json=None, params=None, refresh=None, **kw):
|
def request(self, method, path, json=None, params=None, refresh=None, **kw):
|
||||||
"""Make a request and return its parsed body.
|
"""Make a request and return its parsed body.
|
||||||
|
|||||||
@@ -0,0 +1,105 @@
|
|||||||
|
"""Read captured traffic from a HAR (HTTP Archive) file.
|
||||||
|
|
||||||
|
HAR is the standard export from browser devtools (Chrome, Firefox, Safari) and
|
||||||
|
desktop proxies (Charles, Proxyman). This module turns HAR entries into mimic's
|
||||||
|
flow dicts so the rest of the pipeline (extract, hosts, endpoints, codegen)
|
||||||
|
works unchanged, with no mitmproxy or iPhone setup at all.
|
||||||
|
"""
|
||||||
|
import base64
|
||||||
|
import json
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
|
from . import mitm
|
||||||
|
|
||||||
|
|
||||||
|
def load(path):
|
||||||
|
"""Load a HAR file and return its entries as mimic flow dicts."""
|
||||||
|
with open(path) as f:
|
||||||
|
har = json.load(f)
|
||||||
|
entries = har.get("log", {}).get("entries", [])
|
||||||
|
return [_entry_to_flow(i, e) for i, e in enumerate(entries)]
|
||||||
|
|
||||||
|
|
||||||
|
def _entry_to_flow(index, entry):
|
||||||
|
"""Convert one HAR entry into a mimic flow dict."""
|
||||||
|
req = entry.get("request", {})
|
||||||
|
resp = entry.get("response", {})
|
||||||
|
parsed = urlparse(req.get("url", ""))
|
||||||
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
||||||
|
query = parsed.query
|
||||||
|
return {
|
||||||
|
"id": f"har-{index}",
|
||||||
|
"request": {
|
||||||
|
"host": parsed.hostname,
|
||||||
|
"method": req.get("method"),
|
||||||
|
"path": parsed.path + (f"?{query}" if query else ""),
|
||||||
|
"scheme": parsed.scheme,
|
||||||
|
"port": port,
|
||||||
|
"headers": [[h["name"], h["value"]] for h in req.get("headers", [])],
|
||||||
|
},
|
||||||
|
"response": {
|
||||||
|
"status_code": resp.get("status", 0),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def hosts(path):
|
||||||
|
"""Count requests per host in a HAR file, most frequent first."""
|
||||||
|
return mitm.hosts(load(path))
|
||||||
|
|
||||||
|
|
||||||
|
def endpoints(path, host):
|
||||||
|
"""Distinct (method, path) endpoints for a host, with inline bodies.
|
||||||
|
|
||||||
|
Mirrors mitm.endpoints(), except HAR embeds the request/response bodies in
|
||||||
|
each entry, so there's no separate body fetch. Latest capture of each
|
||||||
|
endpoint wins, and bodies are decoded/truncated the same way as the mitm
|
||||||
|
backend so codegen sees consistent input.
|
||||||
|
"""
|
||||||
|
with open(path) as f:
|
||||||
|
har = json.load(f)
|
||||||
|
by_key = {}
|
||||||
|
for entry in har.get("log", {}).get("entries", []):
|
||||||
|
req = entry.get("request", {})
|
||||||
|
parsed = urlparse(req.get("url", ""))
|
||||||
|
if parsed.hostname != host:
|
||||||
|
continue
|
||||||
|
by_key[(req.get("method"), parsed.path)] = entry
|
||||||
|
|
||||||
|
out = []
|
||||||
|
for (method, path_only), entry in by_key.items():
|
||||||
|
req = entry.get("request", {})
|
||||||
|
resp = entry.get("response", {})
|
||||||
|
out.append(
|
||||||
|
{
|
||||||
|
"method": method,
|
||||||
|
"path": path_only,
|
||||||
|
"status": resp.get("status", 0),
|
||||||
|
"query": urlparse(req.get("url", "")).query,
|
||||||
|
"request_body": mitm._decode(_body_bytes(req.get("postData"))),
|
||||||
|
"response_body": mitm._decode(_content_bytes(resp.get("content"))),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def _body_bytes(post_data):
|
||||||
|
"""Bytes of a HAR request postData block (always plain text)."""
|
||||||
|
if not post_data:
|
||||||
|
return b""
|
||||||
|
return post_data.get("text", "").encode("utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def _content_bytes(content):
|
||||||
|
"""Bytes of a HAR response content block, decoding base64 when flagged."""
|
||||||
|
if not content:
|
||||||
|
return b""
|
||||||
|
text = content.get("text", "")
|
||||||
|
if not text:
|
||||||
|
return b""
|
||||||
|
if content.get("encoding") == "base64":
|
||||||
|
try:
|
||||||
|
return base64.b64decode(text)
|
||||||
|
except ValueError:
|
||||||
|
return text.encode("utf-8")
|
||||||
|
return text.encode("utf-8")
|
||||||
Vendored
+69
@@ -0,0 +1,69 @@
|
|||||||
|
{
|
||||||
|
"log": {
|
||||||
|
"version": "1.2",
|
||||||
|
"creator": {"name": "mimic-test", "version": "1.0"},
|
||||||
|
"entries": [
|
||||||
|
{
|
||||||
|
"request": {
|
||||||
|
"method": "GET",
|
||||||
|
"url": "https://api.example.com/v1/users?page=1",
|
||||||
|
"headers": [
|
||||||
|
{"name": "Authorization", "value": "Bearer token123"},
|
||||||
|
{"name": "User-Agent", "value": "Mozilla/5.0"},
|
||||||
|
{"name": "Host", "value": "api.example.com"}
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"response": {
|
||||||
|
"status": 500,
|
||||||
|
"headers": [],
|
||||||
|
"content": {"mimeType": "application/json", "text": "{\"error\": \"boom\"}"}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"request": {
|
||||||
|
"method": "GET",
|
||||||
|
"url": "https://api.example.com/v1/users?page=2",
|
||||||
|
"headers": [
|
||||||
|
{"name": "Authorization", "value": "Bearer token123"},
|
||||||
|
{"name": "Accept", "value": "application/json"}
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"response": {
|
||||||
|
"status": 200,
|
||||||
|
"headers": [],
|
||||||
|
"content": {"mimeType": "application/json", "text": "{\"users\": []}"}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"request": {
|
||||||
|
"method": "POST",
|
||||||
|
"url": "https://api.example.com/v1/messages",
|
||||||
|
"headers": [
|
||||||
|
{"name": "Authorization", "value": "Bearer token123"},
|
||||||
|
{"name": "Content-Type", "value": "application/json"}
|
||||||
|
],
|
||||||
|
"postData": {"mimeType": "application/json", "text": "{\"to\": \"bob\"}"}
|
||||||
|
},
|
||||||
|
"response": {
|
||||||
|
"status": 201,
|
||||||
|
"headers": [],
|
||||||
|
"content": {"mimeType": "application/json", "encoding": "base64", "text": "eyJvayI6IHRydWV9"}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"request": {
|
||||||
|
"method": "GET",
|
||||||
|
"url": "https://cdn.example.com/img.png",
|
||||||
|
"headers": [
|
||||||
|
{"name": "Accept", "value": "image/png"}
|
||||||
|
]
|
||||||
|
},
|
||||||
|
"response": {
|
||||||
|
"status": 200,
|
||||||
|
"headers": [],
|
||||||
|
"content": {"mimeType": "image/png"}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,70 @@
|
|||||||
|
"""Tests for HAR file import (mimic.sources.har + Session.from_har)."""
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from mimic import Session
|
||||||
|
from mimic.sources import har
|
||||||
|
|
||||||
|
FIXTURE = os.path.join(os.path.dirname(__file__), "fixtures", "sample.har")
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_basic():
|
||||||
|
flows = har.load(FIXTURE)
|
||||||
|
assert len(flows) == 4
|
||||||
|
req = flows[0]["request"]
|
||||||
|
assert req["host"] == "api.example.com"
|
||||||
|
assert req["method"] == "GET"
|
||||||
|
assert req["path"] == "/v1/users?page=1"
|
||||||
|
assert req["scheme"] == "https"
|
||||||
|
assert req["port"] == 443
|
||||||
|
assert req["headers"][0] == ["Authorization", "Bearer token123"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_hosts():
|
||||||
|
assert har.hosts(FIXTURE) == [("api.example.com", 3), ("cdn.example.com", 1)]
|
||||||
|
|
||||||
|
|
||||||
|
def test_endpoints_dedup():
|
||||||
|
eps = har.endpoints(FIXTURE, "api.example.com")
|
||||||
|
assert len(eps) == 2
|
||||||
|
users = [e for e in eps if e["path"] == "/v1/users"]
|
||||||
|
assert len(users) == 1
|
||||||
|
assert users[0]["status"] == 200 # latest (page=2) wins over the earlier 500
|
||||||
|
|
||||||
|
|
||||||
|
def test_session_from_har():
|
||||||
|
s = Session.from_har(FIXTURE, "api.example.com")
|
||||||
|
assert s.base_url == "https://api.example.com"
|
||||||
|
assert s.host == "api.example.com"
|
||||||
|
assert s.headers["Authorization"] == "Bearer token123"
|
||||||
|
|
||||||
|
|
||||||
|
def test_from_har_auto_host():
|
||||||
|
s = Session.from_har(FIXTURE)
|
||||||
|
assert s.host == "api.example.com" # most-requested host in the file
|
||||||
|
|
||||||
|
|
||||||
|
def test_from_har_no_auth(tmp_path):
|
||||||
|
path = tmp_path / "noauth.har"
|
||||||
|
path.write_text(json.dumps({
|
||||||
|
"log": {"entries": [{
|
||||||
|
"request": {
|
||||||
|
"method": "GET",
|
||||||
|
"url": "https://api.example.com/health",
|
||||||
|
"headers": [{"name": "Accept", "value": "application/json"}],
|
||||||
|
},
|
||||||
|
"response": {"status": 200},
|
||||||
|
}]}
|
||||||
|
}))
|
||||||
|
with pytest.raises(RuntimeError):
|
||||||
|
Session.from_har(str(path), "api.example.com")
|
||||||
|
|
||||||
|
|
||||||
|
def test_base64_response_body():
|
||||||
|
eps = har.endpoints(FIXTURE, "api.example.com")
|
||||||
|
msg = [e for e in eps if e["path"] == "/v1/messages"][0]
|
||||||
|
assert '"ok"' in msg["response_body"] # base64 "eyJvayI6IHRydWV9" -> {"ok": true}
|
||||||
|
assert "true" in msg["response_body"]
|
||||||
|
assert '"to"' in msg["request_body"] # postData decoded too
|
||||||
Reference in new issue
Block a user