mirror of
https://github.com/littledivy/mimic.git
synced 2026-10-26 19:15:10 +00:00
mimic: intercept any iOS app, AI-generate a Python client
Capture your own session with mitmproxy, reuse your real auth, and let claude -p write an ergonomic client from the captured endpoints. - Session/App runtime with mitm + cURL sources and token auto-refresh - CLI: record / doctor / hosts / learn / gen - one-command install via uv; mitmproxy launched on demand via uvx - MIT licensed, own-accounts-only framing
This commit is contained in:
commit
3361c2a1d4
12 files changed
+751
No files matched your search
@@ -0,0 +1,6 @@
|
|||||||
|
__pycache__/
|
||||||
|
*_client.py
|
||||||
|
*.egg-info/
|
||||||
|
*.pyc
|
||||||
|
build/
|
||||||
|
dist/
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
MIT License
|
||||||
|
|
||||||
|
Copyright (c) 2026 Divy Srivastava
|
||||||
|
|
||||||
|
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||||
|
of this software and associated documentation files (the "Software"), to deal
|
||||||
|
in the Software without restriction, including without limitation the rights
|
||||||
|
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||||
|
copies of the Software, and to permit persons to whom the Software is
|
||||||
|
furnished to do so, subject to the following conditions:
|
||||||
|
|
||||||
|
The above copyright notice and this permission notice shall be included in all
|
||||||
|
copies or substantial portions of the Software.
|
||||||
|
|
||||||
|
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||||
|
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||||
|
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||||
|
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||||
|
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||||
|
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||||
|
SOFTWARE.
|
||||||
@@ -0,0 +1,104 @@
|
|||||||
|
# mimic
|
||||||
|
|
||||||
|
Intercept any app, then call it from Python like a library.
|
||||||
|
|
||||||
|
```python
|
||||||
|
from hinge_client import Hinge
|
||||||
|
|
||||||
|
acc = Hinge() # reuses your captured session — no keys to wire up
|
||||||
|
recs = acc.get_recommendations()
|
||||||
|
acc.like(subject_id, comment="empty pools are so peaceful lol")
|
||||||
|
```
|
||||||
|
|
||||||
|
You never wrote `hinge_client.py`. mimic watched your own app traffic and an AI
|
||||||
|
wrote the client for you.
|
||||||
|
|
||||||
|
## How it works
|
||||||
|
|
||||||
|
Every app authenticates each request with a stable bundle — a bearer token,
|
||||||
|
device ids, a session id, cookies. Capture that once from a real request **you**
|
||||||
|
made, and you can craft new calls the server can't tell apart from the real app.
|
||||||
|
|
||||||
|
```
|
||||||
|
capture your traffic ─▶ extract your reusable auth ─▶ AI writes the client
|
||||||
|
(mitmproxy) (mimic.Session) (claude reads the
|
||||||
|
captured endpoints)
|
||||||
|
```
|
||||||
|
|
||||||
|
The generated client is plain, editable Python on top of `mimic.App`. It gets
|
||||||
|
the ergonomic parts right — named methods, body templates, and the multi-step
|
||||||
|
chaining mobile APIs need (fetch a token here, spend it there).
|
||||||
|
|
||||||
|
## Install (one command)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
sh install.sh
|
||||||
|
```
|
||||||
|
|
||||||
|
Installs [`uv`](https://astral.sh/uv) if you don't have it, then mimic in an
|
||||||
|
isolated tool env. No separate mitmproxy install — mimic launches it for you via
|
||||||
|
`uvx` on first `record`. (Manual: `uv tool install mimic-client`.)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mimic doctor # confirm proxy + claude are ready
|
||||||
|
```
|
||||||
|
|
||||||
|
## Use it (iPhone)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mimic record # starts the proxy, prints the exact iPhone steps
|
||||||
|
```
|
||||||
|
|
||||||
|
`record` walks you through it and auto-fills your Mac's LAN IP:
|
||||||
|
|
||||||
|
1. iPhone → Wi-Fi → Configure Proxy → Manual → `<your-mac-ip>:8080`
|
||||||
|
2. Safari → `http://mitm.it` → install the Apple profile
|
||||||
|
3. **Settings → General → About → Certificate Trust Settings → turn on full
|
||||||
|
trust for mitmproxy** — the step everyone forgets
|
||||||
|
4. open the app, use it normally
|
||||||
|
|
||||||
|
Then:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
mimic hosts # list captured hosts; pick your API host
|
||||||
|
mimic learn prod-api.hingeaws.net # see the endpoints mimic saw
|
||||||
|
mimic gen prod-api.hingeaws.net # AI writes hinge_client.py
|
||||||
|
```
|
||||||
|
|
||||||
|
Then `from hinge_client import Hinge; Hinge().get_recommendations()`.
|
||||||
|
|
||||||
|
## The library
|
||||||
|
|
||||||
|
Three ways to build a session by hand, if you don't want codegen:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from mimic import Session
|
||||||
|
|
||||||
|
Session.from_mitm("prod-api.hingeaws.net") # pull auth from mitmweb
|
||||||
|
Session.from_curl(open("copied.txt").read()) # paste "Copy as cURL" from devtools
|
||||||
|
Session(base_url="https://x.com", headers={...}) # explicit
|
||||||
|
```
|
||||||
|
|
||||||
|
`.get(path)` / `.post(path, json=...)` return parsed JSON and auto-refresh your
|
||||||
|
token from mitmweb if it rotates (a `401`/`403` triggers one re-pull + retry).
|
||||||
|
|
||||||
|
## Capture backends
|
||||||
|
|
||||||
|
- **mitmproxy** — iOS apps (the default). mimic reads its JSON flow API and runs
|
||||||
|
it for you via `uvx`, so there's nothing extra to install.
|
||||||
|
- **cURL / paste** — anything with a web version. `Copy as cURL` in devtools →
|
||||||
|
`Session.from_curl(text)`. No proxy, no cert.
|
||||||
|
|
||||||
|
**Certificate pinning** (banking, Instagram) defeats a plain proxy, so pinned
|
||||||
|
apps aren't supported. Plenty of apps aren't pinned and just work — if
|
||||||
|
`mimic hosts` shows the app's API host, you're set.
|
||||||
|
|
||||||
|
## Ethics
|
||||||
|
|
||||||
|
Use it on **your own** accounts and data. It replays *your* session; it is not a
|
||||||
|
tool for accessing anyone else's. Respect each app's terms of service.
|
||||||
|
|
||||||
|
## License
|
||||||
|
|
||||||
|
MIT — see [LICENSE](LICENSE). Provided as-is, no warranty. Use on your own
|
||||||
|
accounts and data; you are responsible for complying with each app's terms.
|
||||||
Executable
+21
@@ -0,0 +1,21 @@
|
|||||||
|
#!/usr/bin/env sh
|
||||||
|
# One-command install for mimic. Installs uv if missing, then mimic (isolated).
|
||||||
|
set -e
|
||||||
|
|
||||||
|
if ! command -v uv >/dev/null 2>&1; then
|
||||||
|
echo "installing uv…"
|
||||||
|
curl -LsSf https://astral.sh/uv/install.sh | sh
|
||||||
|
export PATH="$HOME/.local/bin:$PATH"
|
||||||
|
fi
|
||||||
|
|
||||||
|
# Install from the current directory if run inside the repo, else from PyPI.
|
||||||
|
if [ -f "pyproject.toml" ] && grep -q 'name = "mimic-client"' pyproject.toml 2>/dev/null; then
|
||||||
|
uv tool install . --reinstall
|
||||||
|
else
|
||||||
|
uv tool install mimic-client --reinstall
|
||||||
|
fi
|
||||||
|
|
||||||
|
echo
|
||||||
|
echo "installed. next:"
|
||||||
|
echo " mimic doctor # check setup"
|
||||||
|
echo " mimic record # start capture, follow the iPhone steps"
|
||||||
@@ -0,0 +1,12 @@
|
|||||||
|
"""mimic — intercept any app, then call it from Python like a library.
|
||||||
|
|
||||||
|
from mimic import Session, App
|
||||||
|
|
||||||
|
Capture your own session with mitmproxy, and mimic reuses your real auth so the
|
||||||
|
server can't tell your script from the app. Point it at a host and let the AI
|
||||||
|
write an ergonomic client: `mimic gen <host>`.
|
||||||
|
"""
|
||||||
|
from .session import App, Session
|
||||||
|
|
||||||
|
__all__ = ["Session", "App"]
|
||||||
|
__version__ = "0.1.0"
|
||||||
+186
@@ -0,0 +1,186 @@
|
|||||||
|
"""mimic CLI — capture any iOS app, generate a client.
|
||||||
|
|
||||||
|
mimic record start the proxy + print iPhone setup steps
|
||||||
|
mimic hosts list captured hosts (pick your API host here)
|
||||||
|
mimic learn <host> show the endpoints mimic saw for a host
|
||||||
|
mimic gen <host> AI-write a Python client for a host
|
||||||
|
mimic doctor check your setup
|
||||||
|
"""
|
||||||
|
import argparse
|
||||||
|
import re
|
||||||
|
import shutil
|
||||||
|
import socket
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from . import codegen
|
||||||
|
from .sources import mitm
|
||||||
|
|
||||||
|
|
||||||
|
def _mitm_and_flows():
|
||||||
|
m = mitm.Mitm()
|
||||||
|
return m, m.flows()
|
||||||
|
|
||||||
|
|
||||||
|
def _lan_ip():
|
||||||
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||||
|
try:
|
||||||
|
s.connect(("8.8.8.8", 80))
|
||||||
|
return s.getsockname()[0]
|
||||||
|
except OSError:
|
||||||
|
return "<this-machine-ip>"
|
||||||
|
finally:
|
||||||
|
s.close()
|
||||||
|
|
||||||
|
|
||||||
|
def _mitmweb_cmd():
|
||||||
|
"""Prefer mitmweb on PATH; otherwise run it ephemerally through uv."""
|
||||||
|
if shutil.which("mitmweb"):
|
||||||
|
return ["mitmweb"]
|
||||||
|
if shutil.which("uvx"):
|
||||||
|
return ["uvx", "--from", "mitmproxy", "mitmweb"]
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def cmd_record(args):
|
||||||
|
ip = _lan_ip()
|
||||||
|
print(
|
||||||
|
f"""
|
||||||
|
iPhone capture — do this once, then just reopen the app to add traffic:
|
||||||
|
|
||||||
|
1. iPhone → Settings → Wi-Fi → (your network) ⓘ → Configure Proxy → Manual
|
||||||
|
Server: {ip} Port: 8080
|
||||||
|
2. Safari → http://mitm.it → download the Apple (.pem) profile
|
||||||
|
3. Settings → General → VPN & Device Management → install the profile
|
||||||
|
4. Settings → General → About → Certificate Trust Settings
|
||||||
|
→ turn ON full trust for "mitmproxy" ← everyone forgets this
|
||||||
|
5. open the target app and use it normally
|
||||||
|
6. back here: mimic hosts then mimic gen <api-host>
|
||||||
|
|
||||||
|
mitmweb dashboard: http://127.0.0.1:8081 (mimic reads flows from here)
|
||||||
|
|
||||||
|
Some apps (banks, Instagram) pin their certificate, so a proxy sees no
|
||||||
|
usable traffic — those aren't supported. Many apps aren't pinned and just
|
||||||
|
work; if `mimic hosts` shows the app's API host, you're good.
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
cmd = _mitmweb_cmd()
|
||||||
|
if not cmd:
|
||||||
|
sys.exit(
|
||||||
|
"no proxy available — install uv (https://astral.sh/uv) so mimic can\n"
|
||||||
|
"run mitmproxy for you, or `pipx install mitmproxy` yourself."
|
||||||
|
)
|
||||||
|
sys.stdout.flush() # show the steps before the proxy takes over the terminal
|
||||||
|
try:
|
||||||
|
subprocess.run(cmd)
|
||||||
|
except FileNotFoundError:
|
||||||
|
sys.exit("failed to launch mitmweb")
|
||||||
|
|
||||||
|
|
||||||
|
def cmd_doctor(args):
|
||||||
|
ok = True
|
||||||
|
|
||||||
|
def check(name, present, fix):
|
||||||
|
nonlocal ok
|
||||||
|
mark = "ok " if present else "MISSING"
|
||||||
|
print(f" [{mark}] {name}")
|
||||||
|
if not present:
|
||||||
|
ok = False
|
||||||
|
print(f" → {fix}")
|
||||||
|
|
||||||
|
print("mimic setup check:\n")
|
||||||
|
check("proxy (mitmweb or uvx)", _mitmweb_cmd() is not None,
|
||||||
|
"install uv: curl -LsSf https://astral.sh/uv/install.sh | sh")
|
||||||
|
check("claude CLI (for `mimic gen`)", shutil.which("claude") is not None,
|
||||||
|
"install Claude Code, or use `mimic gen --prompt-only`")
|
||||||
|
reachable = False
|
||||||
|
try:
|
||||||
|
mitm.Mitm().flows()
|
||||||
|
reachable = True
|
||||||
|
except mitm.MitmError:
|
||||||
|
pass
|
||||||
|
check("mitmweb running + reachable", reachable,
|
||||||
|
"run `mimic record` in another terminal")
|
||||||
|
print(f"\nLAN IP for the iPhone proxy: {_lan_ip()}:8080")
|
||||||
|
sys.exit(0 if ok else 1)
|
||||||
|
|
||||||
|
|
||||||
|
def cmd_hosts(args):
|
||||||
|
_, flows = _mitm_and_flows()
|
||||||
|
rows = mitm.hosts(flows)
|
||||||
|
if not rows:
|
||||||
|
sys.exit("no traffic captured yet — run `mimic record` and use the app")
|
||||||
|
print(f"{'requests':>9} host")
|
||||||
|
for host, n in rows:
|
||||||
|
print(f"{n:>9} {host}")
|
||||||
|
print("\nPick your API host (usually the one with JSON, not media/cdn).")
|
||||||
|
|
||||||
|
|
||||||
|
def cmd_learn(args):
|
||||||
|
m, flows = _mitm_and_flows()
|
||||||
|
eps = mitm.endpoints(m, flows, args.host)
|
||||||
|
if not eps:
|
||||||
|
sys.exit(f"no requests to {args.host} captured")
|
||||||
|
print(f"{args.host}: {len(eps)} endpoints\n")
|
||||||
|
for e in eps:
|
||||||
|
print(f" {e['method']:5s} {e['path']} -> {e['status']}")
|
||||||
|
|
||||||
|
|
||||||
|
def cmd_gen(args):
|
||||||
|
m, flows = _mitm_and_flows()
|
||||||
|
eps = mitm.endpoints(m, flows, args.host)
|
||||||
|
if not eps:
|
||||||
|
sys.exit(f"no requests to {args.host} captured")
|
||||||
|
|
||||||
|
if args.prompt_only:
|
||||||
|
print(codegen.build_prompt(args.host, eps))
|
||||||
|
return
|
||||||
|
|
||||||
|
out = args.out or _default_out(args.host)
|
||||||
|
print(f"asking claude to write a client from {len(eps)} endpoints…", file=sys.stderr)
|
||||||
|
source = codegen.generate(args.host, eps, model=args.model)
|
||||||
|
with open(out, "w") as f:
|
||||||
|
f.write(source)
|
||||||
|
cls = _class_name(source)
|
||||||
|
print(f"\nwrote {out}")
|
||||||
|
print(f"\n from {out[:-3]} import {cls or 'Client'}")
|
||||||
|
print(f" acc = {cls or 'Client'}()")
|
||||||
|
print(" # then call the generated methods\n")
|
||||||
|
|
||||||
|
|
||||||
|
def _default_out(host):
|
||||||
|
stem = re.sub(r"[^a-z0-9]+", "_", host.split(".")[0].lower()).strip("_")
|
||||||
|
return f"{stem or 'app'}_client.py"
|
||||||
|
|
||||||
|
|
||||||
|
def _class_name(source):
|
||||||
|
m = re.search(r"class\s+(\w+)\s*\(", source)
|
||||||
|
return m.group(1) if m else None
|
||||||
|
|
||||||
|
|
||||||
|
def main(argv=None):
|
||||||
|
p = argparse.ArgumentParser(prog="mimic", description=__doc__,
|
||||||
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||||
|
sub = p.add_subparsers(dest="cmd", required=True)
|
||||||
|
|
||||||
|
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("hosts", help="list captured hosts").set_defaults(func=cmd_hosts)
|
||||||
|
|
||||||
|
lp = sub.add_parser("learn", help="show endpoints for a host")
|
||||||
|
lp.add_argument("host")
|
||||||
|
lp.set_defaults(func=cmd_learn)
|
||||||
|
|
||||||
|
gp = sub.add_parser("gen", help="AI-generate a client for a host")
|
||||||
|
gp.add_argument("host")
|
||||||
|
gp.add_argument("-o", "--out", help="output .py path")
|
||||||
|
gp.add_argument("--model", default="sonnet", help="claude model (default: sonnet)")
|
||||||
|
gp.add_argument("--prompt-only", action="store_true", help="print the prompt instead of calling claude")
|
||||||
|
gp.set_defaults(func=cmd_gen)
|
||||||
|
|
||||||
|
args = p.parse_args(argv)
|
||||||
|
args.func(args)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
@@ -0,0 +1,81 @@
|
|||||||
|
"""Turn captured endpoints into an ergonomic Python client, using the AI.
|
||||||
|
|
||||||
|
`mimic gen <host>` builds a digest of what mimic saw on the wire and pipes it
|
||||||
|
to `claude -p` (Claude Code, headless). Claude writes a real, editable client
|
||||||
|
class — named methods, body templates, response handling, and the multi-step
|
||||||
|
chaining that mobile APIs often need — on top of mimic.App.
|
||||||
|
"""
|
||||||
|
import re
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
|
||||||
|
|
||||||
|
PROMPT = """\
|
||||||
|
You are writing a Python API client. Below is real captured HTTP traffic from \
|
||||||
|
the app `{host}`, recorded by a proxy while the user exercised the app with \
|
||||||
|
their own account. Your job: turn it into a clean, ergonomic client library.
|
||||||
|
|
||||||
|
Rules:
|
||||||
|
- Output ONE Python file, nothing else. No prose, no markdown fences.
|
||||||
|
- Subclass `mimic.App`. Set `HOST = "{host}"`. Auth/device headers are pulled \
|
||||||
|
automatically by the base class — do NOT hardcode tokens or headers.
|
||||||
|
- Give methods human names for what they DO (get_posts, like, send_message), \
|
||||||
|
not the raw path. Infer intent from the path, bodies, and status codes.
|
||||||
|
- Use self.get(path)/self.post(path, json=body). Both return parsed JSON.
|
||||||
|
- If an endpoint's body reuses an id or token that another endpoint returns \
|
||||||
|
(e.g. a viewToken, a playerId, a session id), chain the calls: fetch the \
|
||||||
|
prerequisite inside the method or cache it on the instance. Read the sample \
|
||||||
|
bodies carefully to find these dependencies.
|
||||||
|
- Turn values that vary per call (ids, text, ratings) into method parameters. \
|
||||||
|
Keep values that are constant-for-this-user as defaults or instance state.
|
||||||
|
- Skip pure telemetry/analytics/config endpoints unless they're needed as a \
|
||||||
|
prerequisite for a real action.
|
||||||
|
- Add a one-line docstring per method. Keep it tight and readable.
|
||||||
|
|
||||||
|
Captured endpoints for {host}:
|
||||||
|
|
||||||
|
{digest}
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def build_digest(endpoints):
|
||||||
|
"""Render the endpoint list into the block the AI reads."""
|
||||||
|
parts = []
|
||||||
|
for e in endpoints:
|
||||||
|
block = [f"### {e['method']} {e['path']} -> {e['status']}"]
|
||||||
|
if e["query"]:
|
||||||
|
block.append(f"query: {e['query']}")
|
||||||
|
if e["request_body"]:
|
||||||
|
block.append(f"request body:\n{e['request_body']}")
|
||||||
|
if e["response_body"]:
|
||||||
|
block.append(f"response body:\n{e['response_body']}")
|
||||||
|
parts.append("\n".join(block))
|
||||||
|
return "\n\n".join(parts)
|
||||||
|
|
||||||
|
|
||||||
|
def build_prompt(host, endpoints):
|
||||||
|
return PROMPT.format(host=host, digest=build_digest(endpoints))
|
||||||
|
|
||||||
|
|
||||||
|
def generate(host, endpoints, model="sonnet"):
|
||||||
|
"""Run `claude -p` on the prompt and return the generated Python source."""
|
||||||
|
prompt = build_prompt(host, endpoints)
|
||||||
|
try:
|
||||||
|
proc = subprocess.run(
|
||||||
|
["claude", "-p", "--model", model],
|
||||||
|
input=prompt,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
timeout=300,
|
||||||
|
)
|
||||||
|
except FileNotFoundError:
|
||||||
|
sys.exit("`claude` CLI not found — install Claude Code, or run `mimic gen --prompt-only` and paste the prompt yourself")
|
||||||
|
if proc.returncode != 0:
|
||||||
|
sys.exit(f"claude -p failed:\n{proc.stderr}")
|
||||||
|
return _strip_fences(proc.stdout)
|
||||||
|
|
||||||
|
|
||||||
|
def _strip_fences(text):
|
||||||
|
"""Claude sometimes wraps output in ```python fences despite instructions."""
|
||||||
|
m = re.search(r"```(?:python)?\n(.*?)```", text, re.S)
|
||||||
|
return (m.group(1) if m else text).strip() + "\n"
|
||||||
@@ -0,0 +1,70 @@
|
|||||||
|
"""Pull the reusable part of your captured session out of mitm flows.
|
||||||
|
|
||||||
|
The whole trick behind mimic: an app authenticates each request with a bundle
|
||||||
|
of headers (or cookies) — a bearer token, device ids, a session id. Those are
|
||||||
|
stable across calls. Capture them once from a real request you made, and you
|
||||||
|
can craft new calls that the server can't tell apart from your real client.
|
||||||
|
|
||||||
|
This module finds the newest authenticated request to a host and keeps only the
|
||||||
|
headers worth reusing.
|
||||||
|
"""
|
||||||
|
|
||||||
|
# Header names that carry identity/auth/device context worth replaying.
|
||||||
|
# Anything matching these prefixes or exact names is kept.
|
||||||
|
KEEP_EXACT = {
|
||||||
|
"authorization",
|
||||||
|
"cookie",
|
||||||
|
"user-agent",
|
||||||
|
"accept-language",
|
||||||
|
"content-type",
|
||||||
|
"accept",
|
||||||
|
}
|
||||||
|
KEEP_PREFIX = ("x-",) # x-device-id, x-session-id, x-app-version, x-api-key, …
|
||||||
|
|
||||||
|
# Per-request noise that must NOT be replayed — the HTTP layer sets these.
|
||||||
|
DROP = {
|
||||||
|
"content-length",
|
||||||
|
"host",
|
||||||
|
"connection",
|
||||||
|
"accept-encoding",
|
||||||
|
"content-encoding",
|
||||||
|
"priority",
|
||||||
|
"x-forwarded-for",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _keep(name):
|
||||||
|
n = name.lower()
|
||||||
|
if n in DROP:
|
||||||
|
return False
|
||||||
|
return n in KEEP_EXACT or n.startswith(KEEP_PREFIX)
|
||||||
|
|
||||||
|
|
||||||
|
def session_headers(flows, host):
|
||||||
|
"""Reusable auth/device headers from your newest authed request to `host`.
|
||||||
|
|
||||||
|
"Authed" = has an Authorization or Cookie header. Returns {} if none found.
|
||||||
|
"""
|
||||||
|
for f in reversed(flows):
|
||||||
|
req = f.get("request")
|
||||||
|
if not req or req["host"] != host:
|
||||||
|
continue
|
||||||
|
names = {k.lower() for k, _ in req.get("headers", [])}
|
||||||
|
if not ({"authorization", "cookie"} & names):
|
||||||
|
continue
|
||||||
|
return {k: v for k, v in req["headers"] if _keep(k)}
|
||||||
|
return {}
|
||||||
|
|
||||||
|
|
||||||
|
def base_url(flows, host):
|
||||||
|
"""Reconstruct the scheme://host base from a captured request."""
|
||||||
|
for f in flows:
|
||||||
|
req = f.get("request")
|
||||||
|
if req and req["host"] == host:
|
||||||
|
scheme = req.get("scheme", "https")
|
||||||
|
port = req.get("port")
|
||||||
|
netloc = host
|
||||||
|
if port and port not in (80, 443):
|
||||||
|
netloc = f"{host}:{port}"
|
||||||
|
return f"{scheme}://{netloc}"
|
||||||
|
return f"https://{host}"
|
||||||
@@ -0,0 +1,117 @@
|
|||||||
|
"""The runtime every generated client is built on.
|
||||||
|
|
||||||
|
A Session holds your captured auth and makes calls that look like they came
|
||||||
|
from the real app. Generated clients subclass App and call self.get/.post.
|
||||||
|
|
||||||
|
from hinge_client import Hinge
|
||||||
|
acc = Hinge() # auto-pulls your captured auth from mitmweb
|
||||||
|
acc.get_recs()
|
||||||
|
"""
|
||||||
|
import requests
|
||||||
|
|
||||||
|
from . import extract
|
||||||
|
from .sources.mitm import Mitm
|
||||||
|
|
||||||
|
|
||||||
|
class Session:
|
||||||
|
"""Base URL + reusable headers, with helpers that return parsed JSON.
|
||||||
|
|
||||||
|
Construct one of three ways:
|
||||||
|
Session.from_mitm("prod-api.hingeaws.net") # pull from mitmweb
|
||||||
|
Session(base_url=..., headers={...}) # explicit
|
||||||
|
Session.from_curl(text) # paste a copied cURL
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, base_url, headers=None, host=None, mitm=None):
|
||||||
|
self.base_url = base_url.rstrip("/")
|
||||||
|
self.headers = headers or {}
|
||||||
|
self.host = host
|
||||||
|
self._mitm = mitm # kept for token refresh
|
||||||
|
self._http = requests.Session()
|
||||||
|
|
||||||
|
# ---- constructors -------------------------------------------------------
|
||||||
|
@classmethod
|
||||||
|
def from_mitm(cls, host, mitm=None):
|
||||||
|
mitm = mitm or Mitm()
|
||||||
|
flows = mitm.flows()
|
||||||
|
headers = extract.session_headers(flows, host)
|
||||||
|
if not headers:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"no authenticated request to {host} captured yet — "
|
||||||
|
f"open the app once so mitmweb sees a real call, then retry"
|
||||||
|
)
|
||||||
|
return cls(extract.base_url(flows, host), headers, host=host, mitm=mitm)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_curl(cls, text):
|
||||||
|
base_url, headers = _parse_curl(text)
|
||||||
|
return cls(base_url, headers)
|
||||||
|
|
||||||
|
# ---- calls --------------------------------------------------------------
|
||||||
|
def request(self, method, path, json=None, params=None, refresh=True, **kw):
|
||||||
|
url = path if path.startswith("http") else f"{self.base_url}{path}"
|
||||||
|
r = self._http.request(
|
||||||
|
method, url, headers=self.headers, json=json, params=params, **kw
|
||||||
|
)
|
||||||
|
if r.status_code in (401, 403) and refresh and self.host and self._mitm:
|
||||||
|
# token likely rotated — re-pull from mitm once and retry
|
||||||
|
self.headers = extract.session_headers(self._mitm.flows(), self.host)
|
||||||
|
return self.request(
|
||||||
|
method, path, json=json, params=params, refresh=False, **kw
|
||||||
|
)
|
||||||
|
return _parse(r)
|
||||||
|
|
||||||
|
def get(self, path, **kw):
|
||||||
|
return self.request("GET", path, **kw)
|
||||||
|
|
||||||
|
def post(self, path, json=None, **kw):
|
||||||
|
return self.request("POST", path, json=json, **kw)
|
||||||
|
|
||||||
|
|
||||||
|
class App(Session):
|
||||||
|
"""Subclass this in a generated client and add named methods.
|
||||||
|
|
||||||
|
class Hinge(App):
|
||||||
|
HOST = "prod-api.hingeaws.net"
|
||||||
|
def get_recs(self):
|
||||||
|
return self.post("/rec/v2", {"playerId": self.player_id})
|
||||||
|
"""
|
||||||
|
|
||||||
|
HOST = None
|
||||||
|
|
||||||
|
def __init__(self, **kw):
|
||||||
|
if not kw and self.HOST:
|
||||||
|
got = Session.from_mitm(self.HOST)
|
||||||
|
super().__init__(got.base_url, got.headers, host=got.host, mitm=got._mitm)
|
||||||
|
else:
|
||||||
|
super().__init__(**kw)
|
||||||
|
|
||||||
|
|
||||||
|
def _parse(r):
|
||||||
|
try:
|
||||||
|
return r.json()
|
||||||
|
except ValueError:
|
||||||
|
return r.text
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_curl(text):
|
||||||
|
"""Minimal `curl 'URL' -H 'k: v' ...` parser for the paste fallback."""
|
||||||
|
import shlex
|
||||||
|
from urllib.parse import urlsplit
|
||||||
|
|
||||||
|
tokens = shlex.split(text.replace("\\\n", " "))
|
||||||
|
url, headers = None, {}
|
||||||
|
i = 0
|
||||||
|
while i < len(tokens):
|
||||||
|
t = tokens[i]
|
||||||
|
if t in ("-H", "--header"):
|
||||||
|
i += 1
|
||||||
|
k, _, v = tokens[i].partition(":")
|
||||||
|
headers[k.strip()] = v.strip()
|
||||||
|
elif t.startswith("http"):
|
||||||
|
url = t
|
||||||
|
i += 1
|
||||||
|
if not url:
|
||||||
|
raise ValueError("no URL found in cURL text")
|
||||||
|
parts = urlsplit(url)
|
||||||
|
return f"{parts.scheme}://{parts.netloc}", headers
|
||||||
@@ -0,0 +1 @@
|
|||||||
|
"""Capture backends: mitmproxy, cURL/HAR paste."""
|
||||||
@@ -0,0 +1,110 @@
|
|||||||
|
"""Read captured traffic from a running mitmweb instance.
|
||||||
|
|
||||||
|
mitmweb exposes a JSON API on http://127.0.0.1:8081. Auth is a token passed
|
||||||
|
once as a query param, which sets a session cookie; subsequent requests reuse
|
||||||
|
the cookie. This module pulls the raw flows and normalizes them into a shape
|
||||||
|
that both the runtime Session and the AI codegen step can consume.
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
|
||||||
|
import requests
|
||||||
|
|
||||||
|
DEFAULT_URL = os.environ.get("MITM_URL", "http://127.0.0.1:8081")
|
||||||
|
DEFAULT_TOKEN = os.environ.get("MITM_TOKEN", "test")
|
||||||
|
|
||||||
|
|
||||||
|
class MitmError(RuntimeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class Mitm:
|
||||||
|
"""A thin client over a running mitmweb's flow API."""
|
||||||
|
|
||||||
|
def __init__(self, url=DEFAULT_URL, token=DEFAULT_TOKEN):
|
||||||
|
self.url = url.rstrip("/")
|
||||||
|
self.token = token
|
||||||
|
self._http = requests.Session()
|
||||||
|
|
||||||
|
def _auth(self):
|
||||||
|
try:
|
||||||
|
self._http.get(f"{self.url}/", params={"token": self.token}, timeout=5)
|
||||||
|
except requests.RequestException as e:
|
||||||
|
raise MitmError(
|
||||||
|
f"can't reach mitmweb at {self.url} — is it running? "
|
||||||
|
f"start it with `mitmweb` (original error: {e})"
|
||||||
|
)
|
||||||
|
|
||||||
|
def flows(self):
|
||||||
|
"""All captured flows as a list of dicts (mitmweb's own schema)."""
|
||||||
|
self._auth()
|
||||||
|
r = self._http.get(f"{self.url}/flows", timeout=15)
|
||||||
|
if r.status_code != 200:
|
||||||
|
raise MitmError(f"mitmweb /flows returned {r.status_code}")
|
||||||
|
return r.json()
|
||||||
|
|
||||||
|
def body(self, flow_id, side):
|
||||||
|
"""Raw request/response body bytes for a flow. side is 'request' or 'response'."""
|
||||||
|
r = self._http.get(
|
||||||
|
f"{self.url}/flows/{flow_id}/{side}/content.data", timeout=15
|
||||||
|
)
|
||||||
|
return r.content
|
||||||
|
|
||||||
|
|
||||||
|
def _headers_dict(req):
|
||||||
|
return {k: v for k, v in req.get("headers", [])}
|
||||||
|
|
||||||
|
|
||||||
|
def hosts(flows):
|
||||||
|
"""Count requests per host, most frequent first."""
|
||||||
|
counts = {}
|
||||||
|
for f in flows:
|
||||||
|
req = f.get("request")
|
||||||
|
if req:
|
||||||
|
counts[req["host"]] = counts.get(req["host"], 0) + 1
|
||||||
|
return sorted(counts.items(), key=lambda kv: -kv[1])
|
||||||
|
|
||||||
|
|
||||||
|
def endpoints(mitm, flows, host):
|
||||||
|
"""Distinct (method, path) endpoints for a host, with a sample req/resp body.
|
||||||
|
|
||||||
|
Returns a list of dicts ready to feed to the AI. Latest capture of each
|
||||||
|
endpoint wins, so bodies reflect the most recent real call.
|
||||||
|
"""
|
||||||
|
by_key = {}
|
||||||
|
for f in flows:
|
||||||
|
req = f.get("request")
|
||||||
|
if not req or req["host"] != host:
|
||||||
|
continue
|
||||||
|
by_key[(req["method"], req["path"].split("?")[0])] = f
|
||||||
|
|
||||||
|
out = []
|
||||||
|
for (method, path), f in by_key.items():
|
||||||
|
req = f.get("request") or {}
|
||||||
|
resp = f.get("response") or {}
|
||||||
|
out.append(
|
||||||
|
{
|
||||||
|
"method": method,
|
||||||
|
"path": path,
|
||||||
|
"status": resp.get("status_code"),
|
||||||
|
"query": req["path"].split("?", 1)[1] if "?" in req["path"] else "",
|
||||||
|
"request_body": _decode(mitm.body(f["id"], "request")),
|
||||||
|
"response_body": _decode(mitm.body(f["id"], "response")),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def _decode(raw, limit=4000):
|
||||||
|
"""Best-effort decode of a body to a truncated JSON/text string for the AI."""
|
||||||
|
if not raw:
|
||||||
|
return ""
|
||||||
|
try:
|
||||||
|
text = raw.decode("utf-8")
|
||||||
|
except UnicodeDecodeError:
|
||||||
|
return f"<{len(raw)} bytes binary>"
|
||||||
|
try: # pretty-print + truncate JSON so the AI sees structure, not noise
|
||||||
|
text = json.dumps(json.loads(text), indent=2)
|
||||||
|
except ValueError:
|
||||||
|
pass
|
||||||
|
return text[:limit] + ("\n…(truncated)" if len(text) > limit else "")
|
||||||
@@ -0,0 +1,22 @@
|
|||||||
|
[project]
|
||||||
|
name = "mimic-client"
|
||||||
|
version = "0.1.0"
|
||||||
|
description = "Intercept any app and call it from Python like a library — AI writes the client."
|
||||||
|
requires-python = ">=3.9"
|
||||||
|
license = "MIT"
|
||||||
|
license-files = ["LICENSE"]
|
||||||
|
authors = [{ name = "Divy Srivastava", email = "me@littledivy.com" }]
|
||||||
|
dependencies = ["requests>=2.28"]
|
||||||
|
|
||||||
|
[project.optional-dependencies]
|
||||||
|
capture = ["mitmproxy>=10"]
|
||||||
|
|
||||||
|
[project.scripts]
|
||||||
|
mimic = "mimic.cli:main"
|
||||||
|
|
||||||
|
[build-system]
|
||||||
|
requires = ["setuptools>=64"]
|
||||||
|
build-backend = "setuptools.build_meta"
|
||||||
|
|
||||||
|
[tool.setuptools.packages.find]
|
||||||
|
include = ["mimic*"]
|
||||||
Reference in new issue
Block a user