From 3d86e22008cafe115cf88745ff982ccd8e8f3067 Mon Sep 17 00:00:00 2001 From: subzeroid <143403577+subzeroid@users.noreply.github.com> Date: Sat, 20 Jun 2026 19:26:22 +0300 Subject: [PATCH 1/2] test(aiograpi): live e2e for /info + CI live-test job MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `tests/live/aiograpi_info_e2e.py`: spawns the real `insto` entrypoint (`@instagram -c info --json -`) for a pooled TEST_ACCOUNTS_URL account, performs a full aiograpi login (incl. TOTP, exercising the #45 fix), and asserts the JSON profile (username==instagram, pk==25025320). Reuses each account's client_settings as a seeded session and only honours an explicit IG_PROXY (pooled account proxies are unreliable — 302 on CONNECT). Add offline unit coverage in `tests/test_aiograpi_info_e2e.py` for the pure helpers (skip-clean path, totp extraction, env wiring, session seeding). Wire a `live-test` job into ci.yml (push/workflow_dispatch, canonical repo only) gated on the TEST_ACCOUNTS_URL secret; the script self-skips when the secret is absent. Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/ci.yml | 29 ++++ tests/live/aiograpi_info_e2e.py | 264 ++++++++++++++++++++++++++++++++ tests/test_aiograpi_info_e2e.py | 130 ++++++++++++++++ 3 files changed, 423 insertions(+) create mode 100644 tests/live/aiograpi_info_e2e.py create mode 100644 tests/test_aiograpi_info_e2e.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 32a3c07..e4c17d3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,6 +4,7 @@ on: push: branches: [main] pull_request: + workflow_dispatch: jobs: test: @@ -43,3 +44,31 @@ jobs: # logic lives in mappers + analytics + commands which have 80+%). # See issue tracker before raising back to 80. run: uv run pytest --cov=insto --cov-fail-under=75 + + live-test: + # Live aiograpi end-to-end through the real CLI entrypoint, driven by + # pooled accounts. Canonical-repo only: PRs from forks never receive the + # TEST_ACCOUNTS_URL secret. The script self-skips (exit 0) when the secret + # is absent, so this job is green even before the secret is configured. + runs-on: ubuntu-latest + if: (github.event_name == 'push' || github.event_name == 'workflow_dispatch') && github.repository == 'subzeroid/insto' + env: + TEST_ACCOUNTS_URL: ${{ secrets.TEST_ACCOUNTS_URL }} + IG_PROXY: ${{ secrets.IG_PROXY }} + steps: + - uses: actions/checkout@v6 + + - name: Install uv + uses: astral-sh/setup-uv@v8.1.0 + with: + enable-cache: true + python-version: "3.12" + + - name: Sync dependencies + run: uv sync --extra aiograpi + + - name: aiograpi /info end-to-end + # Spawns `insto @instagram -c info --json -` against a pooled account + # (full login incl. TOTP) and asserts the JSON profile. Skips clean + # if TEST_ACCOUNTS_URL is unset. + run: uv run python tests/live/aiograpi_info_e2e.py diff --git a/tests/live/aiograpi_info_e2e.py b/tests/live/aiograpi_info_e2e.py new file mode 100644 index 0000000..7ed29d6 --- /dev/null +++ b/tests/live/aiograpi_info_e2e.py @@ -0,0 +1,264 @@ +"""Opt-in live end-to-end test for the aiograpi backend, through ``/info``. + +Unlike the saved-feed / private-GraphQL audits (which drive ``AiograpiBackend`` +directly), this script proves the *whole* chain "from install to ``/info``": +it spawns the real ``insto`` entrypoint as a subprocess, configured for the +aiograpi backend via environment variables, logs in with a pooled account +(including TOTP), resolves ``@instagram`` and asserts the JSON profile output. + +Requires ``TEST_ACCOUNTS_URL``. Skips cleanly when unset. The script prints +only status, return codes and shape — never usernames, passwords, proxies, +account URLs, or raw Instagram payloads (all routed through +``insto._redact``). + +Run:: + + TEST_ACCOUNTS_URL=... uv run --extra aiograpi python tests/live/aiograpi_info_e2e.py + +Optional:: + + IG_PROXY=socks5h://127.0.0.1:9050 ... + +Exits non-zero only when accounts cannot be fetched or no pooled account can +complete a correct ``/info`` for ``@instagram``. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path +from typing import Any +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +import httpx + +REPO = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(REPO)) + +from insto._redact import redact_secrets, register_secret # noqa: E402 + +# Stable public target — pk is a fixed Instagram fact, low assertion risk. +TARGET_USERNAME = "instagram" +TARGET_PK = "25025320" +SUBPROCESS_TIMEOUT = 120.0 +# Must match aiograpi's default session filename under INSTO_HOME so the +# subprocess backend picks up the pre-seeded session (see cli/config). +SESSION_FILENAME = "aiograpi.session.json" + + +def _build_accounts_url(raw_url: str, *, count: int) -> str: + parts = urlsplit(raw_url) + query = dict(parse_qsl(parts.query, keep_blank_values=True)) + query["count"] = str(count) + return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment)) + + +def _register_nested_secrets(value: Any) -> None: + if isinstance(value, str): + register_secret(value) + elif isinstance(value, dict): + for nested in value.values(): + _register_nested_secrets(nested) + elif isinstance(value, list | tuple): + for nested in value: + _register_nested_secrets(nested) + + +def _register_account_secrets(accounts_url: str, account: dict[str, Any]) -> None: + register_secret(accounts_url) + for key in ("username", "password", "proxy", "totp_seed"): + register_secret(account.get(key)) + _register_nested_secrets(account.get("client_settings")) + + +def _safe_error(exc: BaseException, *, limit: int = 140) -> str: + first_line = str(exc).splitlines()[0] if str(exc) else "" + message = f"{type(exc).__name__}: {first_line}" + return redact_secrets(message[:limit]) + + +def _account_totp(account: dict[str, Any]) -> str: + """TOTP seed lives either top-level or under ``client_settings``.""" + seed = account.get("totp_seed") + if not seed: + settings = account.get("client_settings") + if isinstance(settings, dict): + seed = settings.get("totp_seed") + return str(seed or "") + + +def _build_subprocess_env( + account: dict[str, Any], + base_env: dict[str, str], + *, + tmp_home: str, +) -> dict[str, str]: + """Construct the child-process env that selects the aiograpi backend. + + ``HIKERAPI_PROXY`` is the proxy the CLI feeds to *every* backend + (``cli._build_backend`` passes ``config.hiker_proxy`` to aiograpi too). + An explicit ``IG_PROXY`` wins over the account's own proxy. + """ + env = dict(base_env) + env["INSTO_HOME"] = tmp_home + env["INSTO_BACKEND"] = "aiograpi" + env["AIOGRAPI_USERNAME"] = str(account.get("username") or "") + env["AIOGRAPI_PASSWORD"] = str(account.get("password") or "") + env["AIOGRAPI_TOTP_SEED"] = _account_totp(account) + # Pooled account proxies are unreliable (frequently 302 on CONNECT, which + # surfaces as AuthRequiredProxyError). Only honour an explicit IG_PROXY, + # matching the saved-feed audit's default behaviour. + proxy = base_env.get("IG_PROXY") + if proxy: + env["HIKERAPI_PROXY"] = str(proxy) + return env + + +def _write_session(tmp_home: str, account: dict[str, Any]) -> str | None: + """Seed the account's saved session so the subprocess reuses its device. + + The pooled accounts ship a ``client_settings`` blob (device uuids, phone + id, etc.). Reusing it — exactly as the saved-feed audit does — lets the + backend validate via ``account_info()`` and skip a cold login that IG + would likely challenge. ``totp_seed`` is never written to the session. + Returns the session path, or ``None`` when the account has no settings. + """ + settings = account.get("client_settings") + if not isinstance(settings, dict) or not settings: + return None + settings = {key: value for key, value in settings.items() if key != "totp_seed"} + path = Path(tmp_home) / SESSION_FILENAME + path.write_text(json.dumps(settings), encoding="utf-8") + os.chmod(path, 0o600) + return str(path) + + +def _extract_profile(stdout: str) -> dict[str, Any]: + """Parse the insto JSON envelope and return its ``data.profile`` object.""" + blob = stdout.strip() + if not blob: + raise RuntimeError("empty stdout from insto subprocess") + try: + envelope = json.loads(blob) + except json.JSONDecodeError: + # Be tolerant of any leading/trailing noise around the JSON object. + start, end = blob.find("{"), blob.rfind("}") + if start == -1 or end == -1: + raise RuntimeError("no JSON object in subprocess stdout") from None + envelope = json.loads(blob[start : end + 1]) + profile = envelope.get("data", {}).get("profile") + if not isinstance(profile, dict): + raise RuntimeError("JSON envelope missing data.profile") + return profile + + +async def _fetch_accounts(accounts_url: str, *, count: int) -> list[dict[str, Any]]: + url = _build_accounts_url(accounts_url, count=count) + register_secret(accounts_url) + register_secret(url) + async with httpx.AsyncClient(timeout=20, verify=False) as client: + response = await client.get(url, headers={"User-Agent": "insto-live-e2e/1"}) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, list): + raise RuntimeError("TEST_ACCOUNTS_URL returned non-list payload") + accounts = [item for item in payload if isinstance(item, dict)] + if not accounts: + raise RuntimeError("TEST_ACCOUNTS_URL returned no account objects") + return accounts + + +async def _info_via_cli(index: int, account: dict[str, Any]) -> bool: + """Run ``insto @instagram -c info --json -`` for one account. True on pass.""" + _register_account_secrets(os.environ.get("TEST_ACCOUNTS_URL", ""), account) + if not account.get("username") or not account.get("password"): + print(f"account[{index}] auth: fail (missing username/password)") + return False + + with tempfile.TemporaryDirectory(prefix=f"insto-e2e-{index}-") as tmp: + env = _build_subprocess_env(account, dict(os.environ), tmp_home=tmp) + _write_session(tmp, account) + cmd = [ + sys.executable, + "-m", + "insto", + f"@{TARGET_USERNAME}", + "-c", + "info", + "--json", + "-", + ] + try: + proc = await asyncio.to_thread( + subprocess.run, + cmd, + env=env, + capture_output=True, + text=True, + timeout=SUBPROCESS_TIMEOUT, + ) + except subprocess.TimeoutExpired: + print(f"account[{index}] info: fail (timeout after {SUBPROCESS_TIMEOUT:.0f}s)") + return False + + if proc.returncode != 0: + tail = (proc.stderr or proc.stdout or "").strip().splitlines() + detail = redact_secrets(tail[-1]) if tail else "no output" + print(f"account[{index}] info: fail (rc={proc.returncode}) {detail[:140]}") + return False + + try: + profile = _extract_profile(proc.stdout) + except Exception as exc: + print(f"account[{index}] info: fail (parse) {_safe_error(exc)}") + return False + + username = str(profile.get("username") or "") + pk = str(profile.get("pk") or "") + if username == TARGET_USERNAME and pk == TARGET_PK: + print(f"account[{index}] info: ok (username={username} pk={pk})") + return True + print(f"account[{index}] info: fail (got username={username!r} pk={pk!r})") + return False + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--count", type=int, default=3, help="number of test accounts to try") + return parser.parse_args(argv) + + +async def main(argv: list[str] | None = None) -> int: + args = _parse_args(list(argv or [])) + accounts_url = os.environ.get("TEST_ACCOUNTS_URL") + if not accounts_url: + print("SKIP: TEST_ACCOUNTS_URL not set") + return 0 + + try: + accounts = await _fetch_accounts(accounts_url, count=args.count) + except Exception as exc: + print(f"FAILED: could not fetch test accounts: {_safe_error(exc)}") + return 1 + + passed = 0 + for index, account in enumerate(accounts[: args.count], start=1): + if await _info_via_cli(index, account): + passed += 1 + break # one clean /info is enough to prove the chain + + if passed: + print(f"\nPASS: /info via aiograpi CLI succeeded ({passed} account)") + return 0 + print(f"\nFAILED: no pooled account completed /info for @{TARGET_USERNAME}") + return 1 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main(sys.argv[1:]))) diff --git a/tests/test_aiograpi_info_e2e.py b/tests/test_aiograpi_info_e2e.py new file mode 100644 index 0000000..dc2f3b5 --- /dev/null +++ b/tests/test_aiograpi_info_e2e.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + +from insto._redact import clear_registered_secrets, register_secret + +MODULE_PATH = Path(__file__).parent / "live" / "aiograpi_info_e2e.py" +SPEC = importlib.util.spec_from_file_location("aiograpi_info_e2e", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +e2e = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = e2e +SPEC.loader.exec_module(e2e) + + +@pytest.fixture(autouse=True) +def _clean_redaction_registry() -> None: + clear_registered_secrets() + yield + clear_registered_secrets() + + +def test_main_skips_clean_when_url_unset( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + import asyncio + + monkeypatch.delenv("TEST_ACCOUNTS_URL", raising=False) + + rc = asyncio.run(e2e.main([])) + + assert rc == 0 + assert "SKIP" in capsys.readouterr().out + + +def test_account_totp_reads_top_level_seed() -> None: + assert e2e._account_totp({"totp_seed": "TOPSEED"}) == "TOPSEED" + + +def test_account_totp_reads_nested_client_settings_seed() -> None: + account = {"client_settings": {"totp_seed": "NESTEDSEED", "uuid": "x"}} + assert e2e._account_totp(account) == "NESTEDSEED" + + +def test_account_totp_missing_returns_empty() -> None: + assert e2e._account_totp({"client_settings": {}}) == "" + + +def test_build_accounts_url_overrides_count() -> None: + url = e2e._build_accounts_url("https://accounts.example.test/list?foo=1&count=9", count=2) + assert url == "https://accounts.example.test/list?foo=1&count=2" + + +def test_safe_error_redacts_registered_values_and_query_tokens() -> None: + register_secret("very-secret-account-url") + exc = RuntimeError( + "failed very-secret-account-url https://api.example.test/path?token=abc123 " + "socks5://user:pass@127.0.0.1:9050" + ) + + out = e2e._safe_error(exc) + + assert "very-secret-account-url" not in out + assert "abc123" not in out + assert "user:pass" not in out + assert "***" in out + + +def test_build_subprocess_env_wires_aiograpi_creds() -> None: + account = { + "username": "ig_user", + "password": "ig_pass", + "proxy": "http://acct-proxy.example:8080", + "client_settings": {"totp_seed": "SEED2FA"}, + } + + env = e2e._build_subprocess_env(account, {"PATH": "/usr/bin"}, tmp_home="/tmp/insto-home") + + assert env["INSTO_BACKEND"] == "aiograpi" + assert env["AIOGRAPI_USERNAME"] == "ig_user" + assert env["AIOGRAPI_PASSWORD"] == "ig_pass" + assert env["AIOGRAPI_TOTP_SEED"] == "SEED2FA" + assert env["INSTO_HOME"] == "/tmp/insto-home" + # Parent env is carried through, not replaced. + assert env["PATH"] == "/usr/bin" + + +def test_build_subprocess_env_ignores_account_proxy_without_ig_proxy() -> None: + # Pooled account proxies are unreliable (often 302 on CONNECT); only an + # explicit IG_PROXY is honoured, matching the saved-feed audit default. + account = { + "username": "u", + "password": "p", + "proxy": "http://acct-proxy.example:8080", + } + + env = e2e._build_subprocess_env(account, {"PATH": "/usr/bin"}, tmp_home="/tmp/h") + + assert "HIKERAPI_PROXY" not in env + + +def test_build_subprocess_env_uses_ig_proxy_env() -> None: + account = {"username": "u", "password": "p", "proxy": "http://acct-proxy.example:8080"} + base = {"PATH": "/usr/bin", "IG_PROXY": "socks5h://127.0.0.1:9050"} + + env = e2e._build_subprocess_env(account, base, tmp_home="/tmp/h") + + assert env["HIKERAPI_PROXY"] == "socks5h://127.0.0.1:9050" + + +def test_write_session_seeds_client_settings_without_totp(tmp_path) -> None: + import json + import stat + + account = {"client_settings": {"uuid": "abc", "totp_seed": "SEED"}} + + path = e2e._write_session(str(tmp_path), account) + + assert path == str(tmp_path / e2e.SESSION_FILENAME) + data = json.loads(Path(path).read_text()) + assert data == {"uuid": "abc"} # totp_seed never persisted to session + assert stat.S_IMODE(Path(path).stat().st_mode) == 0o600 + + +def test_write_session_noop_without_client_settings(tmp_path) -> None: + assert e2e._write_session(str(tmp_path), {"username": "u"}) is None + assert not (tmp_path / e2e.SESSION_FILENAME).exists() From a4b6c556eef0ff57576da84e15580180f926b70c Mon Sep 17 00:00:00 2001 From: subzeroid <143403577+subzeroid@users.noreply.github.com> Date: Sat, 20 Jun 2026 19:28:27 +0300 Subject: [PATCH 2/2] test(aiograpi): enable TLS verification for accounts fetch The accounts host (hikerapi.com) presents a valid certificate, so the verify=False copied from the saved-feed audit is unnecessary. Drop it to address the security review's TLS-verification finding; live e2e still passes with default verification. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/live/aiograpi_info_e2e.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/live/aiograpi_info_e2e.py b/tests/live/aiograpi_info_e2e.py index 7ed29d6..bf59762 100644 --- a/tests/live/aiograpi_info_e2e.py +++ b/tests/live/aiograpi_info_e2e.py @@ -162,7 +162,7 @@ async def _fetch_accounts(accounts_url: str, *, count: int) -> list[dict[str, An url = _build_accounts_url(accounts_url, count=count) register_secret(accounts_url) register_secret(url) - async with httpx.AsyncClient(timeout=20, verify=False) as client: + async with httpx.AsyncClient(timeout=20) as client: response = await client.get(url, headers={"User-Agent": "insto-live-e2e/1"}) response.raise_for_status() payload = response.json()