diff --git a/pypproxy/ab_test/__init__.py b/pypproxy/ab_test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pypproxy/ab_test/runner.py b/pypproxy/ab_test/runner.py new file mode 100644 index 0000000..3c6a4e9 --- /dev/null +++ b/pypproxy/ab_test/runner.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import json +import time +from dataclasses import dataclass + +import httpx + +from pypproxy.store.models import Entry + + +@dataclass +class ABResult: + endpoint_a: str + endpoint_b: str + method: str + status_a: int + status_b: int + body_a: bytes + body_b: bytes + duration_a_ms: int + duration_b_ms: int + error_a: str = "" + error_b: str = "" + + @property + def status_diff(self) -> bool: + return self.status_a != self.status_b + + @property + def body_diff(self) -> bool: + return self.body_a != self.body_b + + def to_dict(self) -> dict: + import base64 + + return { + "endpoint_a": self.endpoint_a, + "endpoint_b": self.endpoint_b, + "method": self.method, + "status_a": self.status_a, + "status_b": self.status_b, + "body_a": base64.b64encode(self.body_a).decode() if self.body_a else "", + "body_b": base64.b64encode(self.body_b).decode() if self.body_b else "", + "duration_a_ms": self.duration_a_ms, + "duration_b_ms": self.duration_b_ms, + "error_a": self.error_a, + "error_b": self.error_b, + "status_diff": self.status_diff, + "body_diff": self.body_diff, + } + + def diff_summary(self) -> str: + lines: list[str] = [] + if self.status_diff: + lines.append(f"Status: A={self.status_a} B={self.status_b}") + else: + lines.append(f"Status: both {self.status_a}") + if self.body_diff: + lines.append(f"Body differs ({len(self.body_a):,} B vs {len(self.body_b):,} B)") + # Try JSON diff summary + try: + da = json.loads(self.body_a) + db = json.loads(self.body_b) + if isinstance(da, dict) and isinstance(db, dict): + added = set(db) - set(da) + removed = set(da) - set(db) + changed = {k for k in da if k in db and da[k] != db[k]} + if added: + lines.append(f" + fields added: {', '.join(sorted(added)[:5])}") + if removed: + lines.append(f" - fields removed: {', '.join(sorted(removed)[:5])}") + if changed: + lines.append(f" ~ fields changed: {', '.join(sorted(changed)[:5])}") + except Exception: + pass + else: + lines.append("Body: identical") + lines.append(f"Latency: A={self.duration_a_ms}ms B={self.duration_b_ms}ms") + return "\n".join(lines) + + +async def run_ab_test( + entry: Entry, + override_host_b: str, + override_scheme_b: str = "", + timeout: int = 30, +) -> ABResult: + """Send the same request to two different hosts and compare responses.""" + scheme = entry.scheme + path = entry.path + query = entry.query + headers = { + k: ", ".join(v) + for k, v in entry.req_headers.items() + if k.lower() not in ("host", "content-length", "connection") + } + body = entry.req_body + + url_a = f"{scheme}://{entry.host}{path}" + (f"?{query}" if query else "") + scheme_b = override_scheme_b or scheme + url_b = f"{scheme_b}://{override_host_b}{path}" + (f"?{query}" if query else "") + + status_a = status_b = 0 + body_a = body_b = b"" + dur_a = dur_b = 0 + err_a = err_b = "" + + async def _fetch(url: str) -> tuple[int, bytes, int, str]: + start = time.monotonic() + try: + h = dict(headers) + h["host"] = url.split("/")[2].split(":")[0] + async with httpx.AsyncClient(verify=False, timeout=timeout, http2=True) as client: + resp = await client.request(method=entry.method, url=url, headers=h, content=body) + return resp.status_code, resp.content, int((time.monotonic() - start) * 1000), "" + except Exception as e: + return 0, b"", int((time.monotonic() - start) * 1000), str(e) + + (status_a, body_a, dur_a, err_a), (status_b, body_b, dur_b, err_b) = await __import__( + "asyncio" + ).gather(_fetch(url_a), _fetch(url_b)) + + return ABResult( + endpoint_a=url_a, + endpoint_b=url_b, + method=entry.method, + status_a=status_a, + status_b=status_b, + body_a=body_a, + body_b=body_b, + duration_a_ms=dur_a, + duration_b_ms=dur_b, + error_a=err_a, + error_b=err_b, + ) diff --git a/pypproxy/api/server.py b/pypproxy/api/server.py index 8401887..0aa0103 100644 --- a/pypproxy/api/server.py +++ b/pypproxy/api/server.py @@ -289,6 +289,89 @@ async def active_scan(req: ScanRequest) -> JSONResponse: return JSONResponse([r.to_dict() for r in results]) +# --- Report --- + + +@app.get("/api/report/html") +async def report_html(host: str = "", title: str = "pypproxy Report") -> PlainTextResponse: + assert _store is not None + from pypproxy.report.generator import generate_html + from pypproxy.store.models import Filter + + f = Filter(host=host) if host else Filter() + entries, _ = _store.list(f, 0, 0) + return PlainTextResponse(generate_html(entries, title), media_type="text/html") + + +@app.get("/api/report/markdown") +async def report_markdown(host: str = "", title: str = "pypproxy Report") -> PlainTextResponse: + assert _store is not None + from pypproxy.report.generator import generate_markdown + from pypproxy.store.models import Filter + + f = Filter(host=host) if host else Filter() + entries, _ = _store.list(f, 0, 0) + return PlainTextResponse(generate_markdown(entries, title), media_type="text/markdown") + + +# --- Macro --- + + +class MacroRunRequest(BaseModel): + steps: list[dict] = [] + timeout: int = 30 + + +@app.post("/api/macro/run") +async def run_macro(req: MacroRunRequest) -> JSONResponse: + from pypproxy.macro.runner import MacroRunner, MacroStep + + steps = [MacroStep.from_dict(s) for s in req.steps] + runner = MacroRunner() + results = await runner.run(steps, timeout=req.timeout) + return JSONResponse([r.to_dict() for r in results]) + + +# --- IDOR --- + + +@app.post("/api/idor") +async def idor_check(data: dict) -> JSONResponse: + assert _store is not None + from pypproxy.security.idor import run_idor_checks + + entry_id = data.get("entry_id") + if not entry_id: + raise HTTPException(status_code=400, detail="entry_id required") + entry = _store.get(entry_id) + if not entry: + raise HTTPException(status_code=404, detail="entry not found") + baseline = data.get("baseline_status", 0) + results = await run_idor_checks(entry, baseline_status=baseline) + return JSONResponse([r.to_dict() for r in results]) + + +# --- A/B Test --- + + +class ABTestRequest(BaseModel): + entry_id: int + host_b: str + scheme_b: str = "" + + +@app.post("/api/ab") +async def ab_test(req: ABTestRequest) -> JSONResponse: + assert _store is not None + from pypproxy.ab_test.runner import run_ab_test + + entry = _store.get(req.entry_id) + if not entry: + raise HTTPException(status_code=404, detail="entry not found") + result = await run_ab_test(entry, req.host_b, req.scheme_b) + return JSONResponse(result.to_dict()) + + # --- GraphQL --- _gql_schema_store = None diff --git a/pypproxy/macro/__init__.py b/pypproxy/macro/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pypproxy/macro/runner.py b/pypproxy/macro/runner.py new file mode 100644 index 0000000..f000576 --- /dev/null +++ b/pypproxy/macro/runner.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import asyncio +import json +import re +import time +from dataclasses import dataclass, field +from typing import Any + +import httpx + +from pypproxy.store.models import Entry + + +@dataclass +class MacroStep: + name: str + method: str + url: str + headers: dict[str, str] = field(default_factory=dict) + body: str = "" + extract: dict[str, str] = field(default_factory=dict) # var_name -> jsonpath-like + assert_status: int = 0 + delay_ms: int = 0 + + def to_dict(self) -> dict: + return { + "name": self.name, + "method": self.method, + "url": self.url, + "headers": self.headers, + "body": self.body, + "extract": self.extract, + "assert_status": self.assert_status, + "delay_ms": self.delay_ms, + } + + @classmethod + def from_dict(cls, d: dict) -> MacroStep: + return cls( + name=d.get("name", ""), + method=d.get("method", "GET"), + url=d.get("url", ""), + headers=d.get("headers", {}), + body=d.get("body", ""), + extract=d.get("extract", {}), + assert_status=d.get("assert_status", 0), + delay_ms=d.get("delay_ms", 0), + ) + + @classmethod + def from_entry(cls, entry: Entry, name: str = "") -> MacroStep: + url = f"{entry.scheme}://{entry.host}{entry.path}" + if entry.query: + url += f"?{entry.query}" + headers = { + k: ", ".join(v) + for k, v in entry.req_headers.items() + if k.lower() not in ("host", "content-length", "connection") + } + body = entry.req_body.decode("utf-8", errors="replace") if entry.req_body else "" + return cls( + name=name or f"{entry.method} {entry.path}", + method=entry.method, + url=url, + headers=headers, + body=body, + ) + + +@dataclass +class StepResult: + step_name: str + status_code: int = 0 + body: str = "" + duration_ms: int = 0 + error: str = "" + extracted: dict[str, str] = field(default_factory=dict) + assertion_passed: bool = True + + def to_dict(self) -> dict: + return { + "step": self.step_name, + "status_code": self.status_code, + "body_preview": self.body[:200], + "duration_ms": self.duration_ms, + "error": self.error, + "extracted": self.extracted, + "assertion_passed": self.assertion_passed, + } + + +class MacroRunner: + def __init__(self) -> None: + self._variables: dict[str, str] = {} + + def set_variable(self, name: str, value: str) -> None: + self._variables[name] = value + + def get_variables(self) -> dict[str, str]: + return dict(self._variables) + + def _substitute(self, text: str) -> str: + """Replace {{var}} placeholders with variable values.""" + + def _replace(m: re.Match) -> str: + return self._variables.get(m.group(1).strip(), m.group(0)) + + return re.sub(r"\{\{(.+?)\}\}", _replace, text) + + def _extract_value(self, body: str, path: str) -> str: + """Extract a value from JSON response body using dot-notation path.""" + try: + data = json.loads(body) + parts = path.strip(".").split(".") + for part in parts: + data = data[int(part)] if isinstance(data, list) else data[part] + return str(data) + except Exception: + return "" + + async def run( + self, + steps: list[MacroStep], + timeout: int = 30, + on_step: Any = None, # callback(step_name, result) + ) -> list[StepResult]: + results: list[StepResult] = [] + + for step in steps: + if step.delay_ms > 0: + await asyncio.sleep(step.delay_ms / 1000) + + url = self._substitute(step.url) + body = self._substitute(step.body) + headers = {k: self._substitute(v) for k, v in step.headers.items()} + + result = StepResult(step_name=step.name) + start = time.monotonic() + + try: + async with httpx.AsyncClient(verify=False, timeout=timeout, http2=True) as client: + resp = await client.request( + method=step.method, + url=url, + headers=headers, + content=body.encode() if body else b"", + ) + result.status_code = resp.status_code + result.duration_ms = int((time.monotonic() - start) * 1000) + try: + result.body = resp.text + except Exception: + result.body = "" + + # Extract variables + for var_name, path in step.extract.items(): + val = self._extract_value(result.body, path) + if val: + self._variables[var_name] = val + result.extracted[var_name] = val + + # Assert status + if step.assert_status and resp.status_code != step.assert_status: + result.assertion_passed = False + result.error = f"Expected {step.assert_status}, got {resp.status_code}" + + except Exception as e: + result.error = str(e) + result.duration_ms = int((time.monotonic() - start) * 1000) + + results.append(result) + + if on_step: + on_step(step.name, result) + + return results + + +def macro_from_entries(entries: list[Entry]) -> list[MacroStep]: + """Convert a list of captured entries into macro steps.""" + return [ + MacroStep.from_entry(e, name=f"Step {i + 1}: {e.method} {e.path}") + for i, e in enumerate(entries) + ] + + +def macro_to_json(steps: list[MacroStep]) -> str: + return json.dumps([s.to_dict() for s in steps], indent=2, ensure_ascii=False) + + +def macro_from_json(data: str) -> list[MacroStep]: + return [MacroStep.from_dict(d) for d in json.loads(data)] diff --git a/pypproxy/report/__init__.py b/pypproxy/report/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pypproxy/report/generator.py b/pypproxy/report/generator.py new file mode 100644 index 0000000..680a2c1 --- /dev/null +++ b/pypproxy/report/generator.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +from pypproxy.store.models import Entry + + +def generate_html( + entries: list[Entry], + title: str = "pypproxy Report", + scan_results: list[dict] | None = None, + security_results: list[dict] | None = None, +) -> str: + now = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC") + hosts = sorted({e.host for e in entries if e.host}) + status_counts: dict[str, int] = {} + for e in entries: + if e.status_code: + bucket = f"{e.status_code // 100}xx" + status_counts[bucket] = status_counts.get(bucket, 0) + 1 + + findings = (scan_results or []) + (security_results or []) + vulns = [f for f in findings if f.get("vulnerable") or f.get("suspicious")] + + rows = [] + for e in entries[:500]: + ct = e.resp_headers.get("content-type", [""])[0][:40] + rows.append(f""" + = 400 else ""}"> + {e.id} + {e.method} + {e.host}{e.path[:60]} + {e.status_code or "—"} + {e.duration_ms or "—"} ms + {len(e.resp_body):,} B + {ct} + """) + + vuln_rows = [] + for f in vulns: + vuln_rows.append(f""" + + {f.get("check", f.get("category", ""))} + {f.get("param", f.get("url", ""))} + {f.get("detail", f.get("description", ""))} + {str(f.get("evidence", f.get("reason", "")))[:100]} + """) + + return f""" + + + +{title} + + + +

{title}

+

Generated {now}  ·  pypproxy

+ +

Summary

+
+
{len(entries)}
Total Requests
+
{len(hosts)}
Hosts
+ {"".join(f'
{v}
{k}
' for k, v in sorted(status_counts.items()))} +
{len(vulns)}
Findings
+
+ +{"

Findings

" + "".join(vuln_rows) + "
CheckTargetDetailEvidence
" if vulns else ""} + +

Traffic ({len(entries)} requests)

+ + + {"".join(rows)} +
#MethodHost + PathStatusTimeSizeContent-Type
+ + + +""" + + +def generate_markdown( + entries: list[Entry], + title: str = "pypproxy Report", + scan_results: list[dict] | None = None, + security_results: list[dict] | None = None, +) -> str: + now = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC") + hosts = sorted({e.host for e in entries if e.host}) + findings = (scan_results or []) + (security_results or []) + vulns = [f for f in findings if f.get("vulnerable") or f.get("suspicious")] + + lines = [ + f"# {title}", + "", + f"Generated: {now} ", + f"Total requests: **{len(entries)}** | Hosts: **{len(hosts)}** | Findings: **{len(vulns)}**", + "", + ] + + if vulns: + lines += ["## Findings", ""] + lines += ["| Check | Target | Detail |", "|-------|--------|--------|"] + for f in vulns: + check = f.get("check", f.get("category", "")) + target = f.get("param", f.get("url", "")) + detail = f.get("detail", f.get("description", "")) + lines.append(f"| {check} | `{target}` | {detail} |") + lines.append("") + + lines += ["## Traffic", ""] + lines += [ + "| # | Method | Host + Path | Status | ms | Size |", + "|---|--------|-------------|--------|----|------|", + ] + for e in entries[:200]: + path = (e.host + e.path)[:60] + lines.append( + f"| {e.id} | {e.method} | `{path}` | {e.status_code or '—'} | {e.duration_ms or '—'} | {len(e.resp_body):,} B |" + ) + + lines += ["", "---", "*Generated by [pypproxy](https://github.com/ykus4/pypproxy)*"] + return "\n".join(lines) diff --git a/pypproxy/security/idor.py b/pypproxy/security/idor.py new file mode 100644 index 0000000..2361076 --- /dev/null +++ b/pypproxy/security/idor.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +import json +import re +import time +from dataclasses import dataclass +from urllib.parse import parse_qs, urlencode + +import httpx + +from pypproxy.store.models import Entry + + +@dataclass +class IDORResult: + param: str + original_value: str + test_value: str + status_code: int = 0 + response_body: bytes = b"" + duration_ms: int = 0 + vulnerable: bool = False + reason: str = "" + + def to_dict(self) -> dict: + import base64 + + return { + "param": self.param, + "original_value": self.original_value, + "test_value": self.test_value, + "status_code": self.status_code, + "body_preview": base64.b64encode(self.response_body[:256]).decode() + if self.response_body + else "", + "duration_ms": self.duration_ms, + "vulnerable": self.vulnerable, + "reason": self.reason, + } + + +_ID_PATTERNS = [ + r"^\d+$", + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", + r"^[0-9a-f]{24}$", # MongoDB ObjectId + r"^[A-Za-z0-9_-]{20,}$", # opaque ID +] + + +def _looks_like_id(value: str) -> bool: + return any(re.match(pat, value, re.IGNORECASE) for pat in _ID_PATTERNS) + + +def _generate_test_ids(original: str) -> list[str]: + tests: list[str] = [] + if re.match(r"^\d+$", original): + n = int(original) + for delta in (1, -1, 2, -2, 0, 99, 100, 1000): + candidate = str(n + delta) + if candidate != original and int(candidate) > 0: + tests.append(candidate) + tests.append("1") + tests.append("2") + elif re.match(r"^[0-9a-f]{8}-", original, re.IGNORECASE): + # UUID: increment last segment + parts = original.split("-") + try: + last = int(parts[-1], 16) + for delta in (1, -1): + new_last = format((last + delta) % (16 ** len(parts[-1])), f"0{len(parts[-1])}x") + tests.append("-".join(parts[:-1] + [new_last])) + except Exception: + pass + else: + tests.extend(["1", "2", "admin", "test", "0"]) + return tests[:6] + + +def _extract_id_params(entry: Entry) -> dict[str, str]: + params: dict[str, str] = {} + # Path segments + for seg in entry.path.split("/"): + if _looks_like_id(seg): + params[f"path:{seg}"] = seg + # Query params + if entry.query: + for k, vs in parse_qs(entry.query).items(): + if vs and _looks_like_id(vs[0]): + params[f"query:{k}"] = vs[0] + # JSON body + if entry.req_body: + try: + data = json.loads(entry.req_body) + if isinstance(data, dict): + for k, v in data.items(): + if isinstance(v, str | int) and _looks_like_id(str(v)): + params[f"body:{k}"] = str(v) + except Exception: + pass + return params + + +def _apply_id(entry: Entry, param_key: str, new_value: str) -> tuple[str, bytes]: + url = f"{entry.scheme}://{entry.host}{entry.path}" + + if param_key.startswith("path:"): + old_seg = param_key[5:] + new_path = entry.path.replace(old_seg, new_value, 1) + url = f"{entry.scheme}://{entry.host}{new_path}" + if entry.query: + url += f"?{entry.query}" + return url, entry.req_body + + if param_key.startswith("query:"): + key = param_key[6:] + qs = parse_qs(entry.query) + qs[key] = [new_value] + url = f"{entry.scheme}://{entry.host}{entry.path}?{urlencode(qs, doseq=True)}" + return url, entry.req_body + + if param_key.startswith("body:"): + key = param_key[5:] + if entry.query: + url += f"?{entry.query}" + try: + data = json.loads(entry.req_body) + data[key] = new_value + return url, json.dumps(data).encode() + except Exception: + return url, entry.req_body + + if entry.query: + url += f"?{entry.query}" + return url, entry.req_body + + +async def run_idor_checks( + entry: Entry, + baseline_status: int = 0, + timeout: int = 15, +) -> list[IDORResult]: + id_params = _extract_id_params(entry) + if not id_params: + return [] + + req_headers = { + k: ", ".join(v) + for k, v in entry.req_headers.items() + if k.lower() not in ("host", "content-length") + } + + results: list[IDORResult] = [] + + for param_key, original_value in id_params.items(): + for test_value in _generate_test_ids(original_value): + url, body = _apply_id(entry, param_key, test_value) + start = time.monotonic() + try: + async with httpx.AsyncClient(verify=False, timeout=timeout, http2=True) as client: + resp = await client.request( + method=entry.method, + url=url, + headers=req_headers, + content=body, + ) + status = resp.status_code + resp_body = resp.content[:512] + dur = int((time.monotonic() - start) * 1000) + + # Vulnerable: different ID returns 2xx (possible IDOR) + vulnerable = ( + status in range(200, 300) + and ( + baseline_status in (401, 403, 404, 0) or baseline_status in range(200, 300) + ) + and test_value != original_value + ) + reason = "" + if vulnerable and baseline_status in (401, 403): + reason = f"Returns {status} with different ID (baseline was {baseline_status})" + elif vulnerable: + reason = f"Returns {status} with different ID {test_value!r}" + + results.append( + IDORResult( + param=param_key, + original_value=original_value, + test_value=test_value, + status_code=status, + response_body=resp_body, + duration_ms=dur, + vulnerable=vulnerable, + reason=reason, + ) + ) + except Exception as e: + results.append( + IDORResult( + param=param_key, + original_value=original_value, + test_value=test_value, + duration_ms=int((time.monotonic() - start) * 1000), + reason=str(e), + ) + ) + + return results diff --git a/pypproxy/session/__init__.py b/pypproxy/session/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pypproxy/session/manager.py b/pypproxy/session/manager.py new file mode 100644 index 0000000..a6792aa --- /dev/null +++ b/pypproxy/session/manager.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import json +import threading +from dataclasses import dataclass, field +from datetime import UTC, datetime +from pathlib import Path + + +@dataclass +class Session: + id: str + name: str + created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) + entry_ids: list[int] = field(default_factory=list) + notes: str = "" + tags: list[str] = field(default_factory=list) + + def to_dict(self) -> dict: + return { + "id": self.id, + "name": self.name, + "created_at": self.created_at, + "entry_ids": self.entry_ids, + "notes": self.notes, + "tags": self.tags, + } + + @classmethod + def from_dict(cls, d: dict) -> Session: + return cls( + id=d["id"], + name=d["name"], + created_at=d.get("created_at", ""), + entry_ids=d.get("entry_ids", []), + notes=d.get("notes", ""), + tags=d.get("tags", []), + ) + + +class SessionManager: + def __init__(self, sessions_dir: str = "") -> None: + self._sessions: dict[str, Session] = {} + self._active_id: str | None = None + self._lock = threading.Lock() + self._sessions_dir = ( + Path(sessions_dir) if sessions_dir else Path.home() / ".pypproxy" / "sessions" + ) + self._sessions_dir.mkdir(parents=True, exist_ok=True) + self._load_all() + + # --- CRUD --- + + def create(self, name: str, notes: str = "") -> Session: + import uuid + + sess = Session(id=str(uuid.uuid4())[:8], name=name, notes=notes) + with self._lock: + self._sessions[sess.id] = sess + self._save(sess) + return sess + + def get(self, session_id: str) -> Session | None: + with self._lock: + return self._sessions.get(session_id) + + def list(self) -> list[Session]: + with self._lock: + return sorted(self._sessions.values(), key=lambda s: s.created_at, reverse=True) + + def delete(self, session_id: str) -> None: + with self._lock: + self._sessions.pop(session_id, None) + if self._active_id == session_id: + self._active_id = None + path = self._sessions_dir / f"{session_id}.json" + if path.exists(): + path.unlink() + + def rename(self, session_id: str, name: str) -> None: + with self._lock: + sess = self._sessions.get(session_id) + if sess: + sess.name = name + self._save(sess) + + def update_notes(self, session_id: str, notes: str) -> None: + with self._lock: + sess = self._sessions.get(session_id) + if sess: + sess.notes = notes + self._save(sess) + + # --- Active session --- + + @property + def active_id(self) -> str | None: + return self._active_id + + def set_active(self, session_id: str | None) -> None: + self._active_id = session_id + + def get_active(self) -> Session | None: + if self._active_id: + return self.get(self._active_id) + return None + + # --- Entry management --- + + def add_entry(self, session_id: str, entry_id: int) -> None: + with self._lock: + sess = self._sessions.get(session_id) + if sess and entry_id not in sess.entry_ids: + sess.entry_ids.append(entry_id) + self._save(sess) + + def remove_entry(self, session_id: str, entry_id: int) -> None: + with self._lock: + sess = self._sessions.get(session_id) + if sess and entry_id in sess.entry_ids: + sess.entry_ids.remove(entry_id) + self._save(sess) + + # --- Persistence --- + + def _save(self, sess: Session) -> None: + path = self._sessions_dir / f"{sess.id}.json" + with open(path, "w") as f: + json.dump(sess.to_dict(), f, indent=2) + + def _load_all(self) -> None: + for path in self._sessions_dir.glob("*.json"): + try: + with open(path) as f: + data = json.load(f) + sess = Session.from_dict(data) + self._sessions[sess.id] = sess + except Exception: + pass diff --git a/pypproxy/ui/ab_tab.py b/pypproxy/ui/ab_tab.py new file mode 100644 index 0000000..d306c49 --- /dev/null +++ b/pypproxy/ui/ab_tab.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import json + +from nicegui import ui + +from pypproxy.store.models import Entry +from pypproxy.store.store import Store + + +def build_ab_tab(store: Store) -> dict: + state: dict = {"entry": None} + + with ui.column().classes("w-full h-full overflow-auto q-pa-md"): + ui.label("A/B Test").classes("text-subtitle2 q-mb-xs") + ui.label("Send the same request to two endpoints and compare responses.").classes( + "text-caption text-grey q-mb-md" + ) + + entry_label = ui.label("No entry selected").classes("text-grey text-caption q-mb-sm") + + with ui.row().classes("gap-2 items-end q-mb-sm flex-wrap"): + host_a = ( + ui.input(label="Host A (from entry)").props("dense outlined dark").classes("w-56") + ) + host_b = ( + ui.input(label="Host B (override)").props("dense outlined dark").classes("w-56") + ) + scheme_b = ( + ui.select(["", "http", "https"], value="", label="Scheme B") + .props("dense outlined dark") + .classes("w-24") + ) + run_btn = ui.button("Run A/B", icon="compare").props("color=primary") + + summary_label = ui.label("").classes("text-caption text-grey q-mb-sm") + + with ui.row().classes("gap-2 w-full"): + with ui.column().classes("flex-1"): + ui.label("Host A").classes("text-caption text-weight-bold") + status_a = ui.label("").classes("text-caption") + body_a = ( + ui.textarea() + .props("outlined dense rows=14 readonly") + .classes("w-full font-mono text-xs") + ) + + with ui.column().classes("flex-1"): + ui.label("Host B").classes("text-caption text-weight-bold") + status_b = ui.label("").classes("text-caption") + body_b = ( + ui.textarea() + .props("outlined dense rows=14 readonly") + .classes("w-full font-mono text-xs") + ) + + ui.label("Diff summary:").classes("text-caption text-weight-bold q-mt-sm") + diff_area = ( + ui.textarea() + .props("outlined dense rows=8 readonly") + .classes("w-full font-mono text-xs") + ) + + async def _run() -> None: + entry = state.get("entry") + if not entry: + ui.notify("Select an entry from Traffic first", type="warning") + return + b_host = host_b.value.strip() + if not b_host: + ui.notify("Enter Host B", type="warning") + return + + run_btn.props("loading") + try: + from pypproxy.ab_test.runner import run_ab_test + + result = await run_ab_test(entry, b_host, scheme_b.value or "") + + status_a.text = f"Status: {result.status_a} ({result.duration_a_ms} ms)" + status_b.text = f"Status: {result.status_b} ({result.duration_b_ms} ms)" + + def _fmt(b: bytes) -> str: + try: + parsed = json.loads(b) + return json.dumps(parsed, indent=2, ensure_ascii=False) + except Exception: + return b.decode("utf-8", errors="replace") + + text_a = _fmt(result.body_a) + text_b = _fmt(result.body_b) + body_a.value = text_a + body_b.value = text_b + + diff_area.value = result.diff_summary() + + summary_label.text = ( + f"Status diff: {'YES ⚠' if result.status_diff else 'no'} | " + f"Body diff: {'YES ⚠' if result.body_diff else 'no'}" + ) + + if result.status_diff or result.body_diff: + ui.notify("Differences found!", type="warning") + else: + ui.notify("Responses are identical", type="positive") + finally: + run_btn.props(remove="loading") + + run_btn.on("click", _run) + + def open_entry(entry: Entry) -> None: + state["entry"] = entry + entry_label.text = f"#{entry.id} {entry.method} {entry.scheme}://{entry.host}{entry.path}" + host_a.value = entry.host + + return {"open_entry": open_entry} diff --git a/pypproxy/ui/app.py b/pypproxy/ui/app.py index 80e2619..c810c18 100644 --- a/pypproxy/ui/app.py +++ b/pypproxy/ui/app.py @@ -83,6 +83,10 @@ async def index() -> None: graphql_tab = ui.tab("GraphQL", icon="account_tree") codegen_tab = ui.tab("Code Gen", icon="code") frida_tab = ui.tab("Frida", icon="bug_report") + macro_tab = ui.tab("Macro", icon="playlist_play") + ab_tab = ui.tab("A/B Test", icon="compare") + session_tab = ui.tab("Sessions", icon="folder") + report_tab = ui.tab("Report", icon="summarize") openapi_tab_btn = ui.tab("OpenAPI", icon="description") analytics_tab_btn = ui.tab("Analytics", icon="bar_chart") import_tab_btn = ui.tab("Import/Search", icon="upload") @@ -163,6 +167,26 @@ async def index() -> None: frida_state = build_frida_tab(store) + with ui.tab_panel(macro_tab).classes("p-0 h-full"): + from .macro_tab import build_macro_tab + + macro_state = build_macro_tab(store) + + with ui.tab_panel(ab_tab).classes("p-0 h-full"): + from .ab_tab import build_ab_tab + + ab_state = build_ab_tab(store) + + with ui.tab_panel(session_tab).classes("p-0 h-full"): + from .session_tab import build_session_tab + + session_state = build_session_tab(store) + + with ui.tab_panel(report_tab).classes("p-0 h-full"): + from .report_tab import build_report_tab + + build_report_tab(store) + with ui.tab_panel(import_tab_btn).classes("p-0 h-full"): from .import_tab import build_import_tab @@ -237,6 +261,21 @@ async def on_row_contextmenu(e) -> None: # noqa: ANN001 tabs.set_value(codegen_tab), ), ) + ui.menu_item( + "A/B Test", + on_click=lambda: (ab_state["open_entry"](entry), tabs.set_value(ab_tab)), + ) + ui.menu_item( + "Add to Macro", + on_click=lambda: ( + macro_state["open_entries"]([entry]), + tabs.set_value(macro_tab), + ), + ) + ui.menu_item( + "Add to Session", + on_click=lambda: session_state["add_entry_to_active"](entry), + ) if "graphql" in (entry.tags or []): ui.menu_item( "Open in GraphQL tab", diff --git a/pypproxy/ui/macro_tab.py b/pypproxy/ui/macro_tab.py new file mode 100644 index 0000000..120a7d0 --- /dev/null +++ b/pypproxy/ui/macro_tab.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import json + +from nicegui import ui + +from pypproxy.store.models import Entry +from pypproxy.store.store import Store + + +def build_macro_tab(store: Store) -> dict: + with ui.column().classes("w-full h-full overflow-auto q-pa-md"): + ui.label("Macro Runner").classes("text-subtitle2 q-mb-xs") + ui.label( + "Chain multiple requests in sequence. Use {{var}} placeholders — values are extracted from responses." + ).classes("text-caption text-grey q-mb-sm") + + with ui.row().classes("gap-2 items-center q-mb-sm"): + run_btn = ui.button("Run Macro", icon="play_arrow").props("color=primary") + clear_btn = ui.button("Clear Steps", icon="delete_sweep").props( + "flat color=negative size=sm" + ) + ui.button("Import JSON", icon="upload").props("flat size=sm") + export_btn = ui.button("Export JSON", icon="download").props("flat size=sm") + + steps_label = ui.label("0 steps").classes("text-caption text-grey") + + # Steps editor + ui.label("Steps (JSON):").classes("text-caption text-weight-bold q-mt-sm") + steps_area = ( + ui.textarea( + placeholder='[\n {\n "name": "Login",\n "method": "POST",\n "url": "https://api.example.com/login",\n "body": "{\\"username\\":\\"alice\\",\\"password\\":\\"secret\\"}",\n "extract": {"token": "data.access_token"},\n "assert_status": 200\n },\n {\n "name": "Get profile",\n "method": "GET",\n "url": "https://api.example.com/profile",\n "headers": {"Authorization": "Bearer {{token}}"},\n "assert_status": 200\n }\n]' + ) + .props("outlined dense rows=14") + .classes("w-full font-mono text-xs") + ) + + # Variables display + ui.label("Extracted variables:").classes("text-caption text-weight-bold q-mt-sm") + vars_label = ui.label("(none yet)").classes("text-caption text-grey font-mono q-mb-sm") + + # Results + ui.label("Results:").classes("text-caption text-weight-bold") + results_table = ( + ui.table( + columns=[ + {"name": "step", "label": "Step", "field": "step", "align": "left"}, + { + "name": "status", + "label": "Status", + "field": "status_code", + "align": "center", + }, + {"name": "ms", "label": "ms", "field": "duration_ms", "align": "right"}, + { + "name": "ok", + "label": "Assert", + "field": "assertion_passed", + "align": "center", + }, + { + "name": "extracted", + "label": "Extracted", + "field": "extracted_str", + "align": "left", + }, + {"name": "error", "label": "Error", "field": "error", "align": "left"}, + ], + rows=[], + row_key="step", + ) + .classes("w-full") + .props("dense flat dark") + ) + results_table.add_slot( + "body-cell-ok", + """ + + + + """, + ) + results_table.add_slot( + "body-cell-status", + """ + + + + """, + ) + + async def _run_macro() -> None: + from pypproxy.macro.runner import MacroRunner, macro_from_json + + try: + steps = macro_from_json(steps_area.value) + except Exception as e: + ui.notify(f"Invalid JSON: {e}", type="negative") + return + if not steps: + ui.notify("No steps to run", type="warning") + return + + run_btn.props("loading") + steps_label.text = f"Running {len(steps)} steps…" + results_table.rows = [] + results_table.update() + + try: + runner = MacroRunner() + results = await runner.run(steps) + rows = [] + for r in results: + rows.append( + { + "step": r.step_name, + "status_code": r.status_code, + "duration_ms": r.duration_ms, + "assertion_passed": r.assertion_passed, + "extracted_str": ", ".join( + f"{k}={v[:20]}" for k, v in r.extracted.items() + ), + "error": r.error[:60] if r.error else "", + } + ) + results_table.rows = rows + results_table.update() + + # Show extracted vars + all_extracted: dict[str, str] = {} + for r in results: + all_extracted.update(r.extracted) + vars_label.text = ( + json.dumps(all_extracted, ensure_ascii=False) if all_extracted else "(none)" + ) + + fails = sum(1 for r in results if not r.assertion_passed or r.error) + steps_label.text = f"{len(steps)} steps — {fails} failed" + if fails: + ui.notify(f"{fails} step(s) failed", type="warning") + else: + ui.notify("All steps passed", type="positive") + finally: + run_btn.props(remove="loading") + + def _clear() -> None: + steps_area.value = "" + results_table.rows = [] + results_table.update() + vars_label.text = "(none yet)" + steps_label.text = "0 steps" + + def _export() -> None: + if steps_area.value.strip(): + ui.download(steps_area.value.encode(), "macro.json") + + run_btn.on("click", _run_macro) + clear_btn.on("click", _clear) + export_btn.on("click", _export) + + def open_entries(entries: list[Entry]) -> None: + from pypproxy.macro.runner import macro_from_entries, macro_to_json + + steps = macro_from_entries(entries) + steps_area.value = macro_to_json(steps) + steps_label.text = f"{len(steps)} steps loaded" + ui.notify(f"Loaded {len(steps)} steps from traffic", type="info") + + return {"open_entries": open_entries} diff --git a/pypproxy/ui/report_tab.py b/pypproxy/ui/report_tab.py new file mode 100644 index 0000000..1afbbdc --- /dev/null +++ b/pypproxy/ui/report_tab.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from nicegui import ui + +from pypproxy.store.models import Filter +from pypproxy.store.store import Store + + +def build_report_tab(store: Store) -> None: + with ui.column().classes("w-full h-full overflow-auto q-pa-md"): + ui.label("Report Generator").classes("text-subtitle2 q-mb-xs") + ui.label("Export captured traffic and findings as HTML or Markdown.").classes( + "text-caption text-grey q-mb-md" + ) + + with ui.row().classes("gap-4 items-end q-mb-md flex-wrap"): + title_input = ( + ui.input(label="Report title", value="pypproxy Report") + .props("dense outlined dark") + .classes("w-64") + ) + fmt_select = ( + ui.select(["HTML", "Markdown"], value="HTML", label="Format") + .props("dense outlined dark") + .classes("w-28") + ) + host_filter = ( + ui.input(label="Filter host (optional)") + .props("dense outlined dark") + .classes("w-48") + ) + gen_btn = ui.button("Generate", icon="description").props("color=primary") + + summary_label = ui.label("").classes("text-caption text-grey q-mb-xs") + report_area = ( + ui.textarea() + .props("outlined dense rows=28 readonly") + .classes("w-full font-mono text-xs") + ) + + with ui.row().classes("gap-2"): + ui.button( + "Copy", + icon="content_copy", + on_click=lambda: ( + ui.run_javascript(f"navigator.clipboard.writeText({report_area.value!r})"), + ui.notify("Copied!", type="positive"), + ), + ).props("flat size=sm") + ui.button( + "Download", + icon="download", + on_click=lambda: ui.download( + report_area.value.encode(), + f"report.{'html' if fmt_select.value == 'HTML' else 'md'}", + ), + ).props("flat size=sm") + + def _generate() -> None: + from pypproxy.report.generator import generate_html, generate_markdown + + f = Filter(host=host_filter.value.strip()) if host_filter.value.strip() else Filter() + entries, total = store.list(f, 0, 0) + if not entries: + ui.notify("No entries", type="warning") + return + title = title_input.value or "pypproxy Report" + if fmt_select.value == "HTML": + report_area.value = generate_html(entries, title) + else: + report_area.value = generate_markdown(entries, title) + summary_label.text = f"Generated from {total} entries" + ui.notify(f"Report ready ({total} requests)", type="positive") + + gen_btn.on("click", _generate) diff --git a/pypproxy/ui/session_tab.py b/pypproxy/ui/session_tab.py new file mode 100644 index 0000000..3a8bd16 --- /dev/null +++ b/pypproxy/ui/session_tab.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from nicegui import ui + +from pypproxy.session.manager import SessionManager +from pypproxy.store.models import Entry +from pypproxy.store.store import Store + +_session_mgr: SessionManager | None = None + + +def get_session_manager() -> SessionManager: + global _session_mgr + if _session_mgr is None: + _session_mgr = SessionManager() + return _session_mgr + + +def build_session_tab(store: Store) -> dict: + mgr = get_session_manager() + + with ui.column().classes("w-full h-full overflow-auto q-pa-md"): + ui.label("Sessions").classes("text-subtitle2 q-mb-xs") + ui.label("Group and save entries into named sessions.").classes( + "text-caption text-grey q-mb-sm" + ) + + # Create new session + with ui.row().classes("gap-2 items-center q-mb-md"): + name_input = ( + ui.input(label="Session name", placeholder="e.g. Login flow") + .props("dense outlined dark") + .classes("w-56") + ) + create_btn = ui.button("New Session", icon="add").props("color=primary size=sm") + + sessions_container = ui.column().classes("w-full") + + def _refresh_sessions() -> None: + sessions_container.clear() + with sessions_container: + sessions = mgr.list() + if not sessions: + ui.label("No sessions yet").classes("text-grey text-caption") + return + for sess in sessions: + active = mgr.active_id == sess.id + with ( + ui.card().classes(f"w-full q-mb-xs {'bg-blue-grey-10' if active else ''}"), + ui.row().classes("items-center gap-2"), + ): + if active: + ui.badge("active", color="primary").props("rounded") + ui.label(sess.name).classes("text-weight-medium flex-1") + ui.label(f"{len(sess.entry_ids)} entries").classes("text-caption text-grey") + ui.button( + icon="radio_button_checked" if active else "radio_button_unchecked", + on_click=lambda sid=sess.id: ( + _set_active(sid), + _refresh_sessions(), + ), + ).props("flat dense size=sm color=primary").tooltip("Set active") + ui.button( + icon="delete", + on_click=lambda sid=sess.id: (mgr.delete(sid), _refresh_sessions()), + ).props("flat dense size=sm color=negative") + + def _set_active(session_id: str) -> None: + if mgr.active_id == session_id: + mgr.set_active(None) + else: + mgr.set_active(session_id) + + def _create_session() -> None: + name = name_input.value.strip() + if not name: + ui.notify("Enter a name", type="warning") + return + mgr.create(name) + name_input.value = "" + _refresh_sessions() + ui.notify(f"Session '{name}' created", type="positive") + + create_btn.on("click", _create_session) + _refresh_sessions() + + def add_entry_to_active(entry: Entry) -> None: + active = mgr.get_active() + if active: + mgr.add_entry(active.id, entry.id) + ui.notify(f"Added to session '{active.name}'", type="positive") + else: + ui.notify("No active session — create or activate one first", type="warning") + + return {"add_entry_to_active": add_entry_to_active, "manager": mgr} diff --git a/tests/test_all_features.py b/tests/test_all_features.py new file mode 100644 index 0000000..b509de8 --- /dev/null +++ b/tests/test_all_features.py @@ -0,0 +1,306 @@ +from __future__ import annotations + +import json + +import pytest + +from pypproxy.ab_test.runner import ABResult +from pypproxy.macro.runner import ( + MacroStep, + macro_from_entries, + macro_from_json, + macro_to_json, +) +from pypproxy.report.generator import generate_html, generate_markdown +from pypproxy.security.idor import _extract_id_params, _generate_test_ids +from pypproxy.session.manager import SessionManager +from pypproxy.store.models import Entry + + +def make_entry(**kwargs) -> Entry: + e = Entry( + method="GET", + scheme="https", + host="api.example.com", + path="/v1/users/123", + status_code=200, + protocol="https", + ) + for k, v in kwargs.items(): + setattr(e, k, v) + e.id = kwargs.get("id", 1) + return e + + +# ---- Report ---- + + +def test_generate_html_basic(): + entries = [make_entry(), make_entry(id=2, host="other.com")] + html = generate_html(entries, "Test Report") + assert " 0 + + +def test_extract_id_params_query(): + e = make_entry(path="/search", query="user_id=789&page=1") + params = _extract_id_params(e) + assert "query:user_id" in params + + +def test_extract_id_params_json_body(): + e = make_entry(method="POST", path="/delete") + e.req_body = json.dumps({"item_id": "42", "reason": "test"}).encode() + params = _extract_id_params(e) + assert "body:item_id" in params + + +def test_generate_test_ids_numeric(): + ids = _generate_test_ids("100") + assert "101" in ids + assert "99" in ids + assert len(ids) >= 4 + + +def test_generate_test_ids_uuid(): + uuid = "550e8400-e29b-41d4-a716-446655440000" + ids = _generate_test_ids(uuid) + assert len(ids) > 0 + assert uuid not in ids + + +# ---- A/B Test result ---- + + +def test_ab_result_diff_detection(): + result = ABResult( + endpoint_a="https://a.com/api", + endpoint_b="https://b.com/api", + method="GET", + status_a=200, + status_b=403, + body_a=b'{"data":"a"}', + body_b=b'{"error":"forbidden"}', + duration_a_ms=100, + duration_b_ms=200, + ) + assert result.status_diff + assert result.body_diff + + +def test_ab_result_no_diff(): + body = b'{"ok": true}' + result = ABResult( + endpoint_a="https://a.com", + endpoint_b="https://b.com", + method="GET", + status_a=200, + status_b=200, + body_a=body, + body_b=body, + duration_a_ms=50, + duration_b_ms=60, + ) + assert not result.status_diff + assert not result.body_diff + + +def test_ab_result_diff_summary(): + result = ABResult( + endpoint_a="https://a.com", + endpoint_b="https://b.com", + method="GET", + status_a=200, + status_b=404, + body_a=b"ok", + body_b=b"not found", + duration_a_ms=10, + duration_b_ms=20, + ) + summary = result.diff_summary() + assert "Status" in summary + assert "200" in summary + assert "404" in summary + + +@pytest.mark.asyncio +async def test_ab_run_unreachable(): + from pypproxy.ab_test.runner import run_ab_test + + e = make_entry(scheme="http", host="127.0.0.1:1") + result = await run_ab_test(e, "127.0.0.1:2", timeout=2) + assert result.error_a != "" or result.error_b != ""