feat: add safe login profile research tool

.gitignore

@@ -45,6 +45,7 @@ pip-delete-this-directory.txt
 
 # Unit test / coverage reports
 htmlcov/
+login-matrix*.jsonl
 .tox/
 .nox/
 .coverage

docs/development-guide.md

@@ -109,6 +109,24 @@ publishes versioned docs with `mike`.
 The `Upstream Sync Tracker` workflow can be triggered manually or by repository dispatch when `instagrapi` publishes a
 new release. It creates a tracking issue with the current async-port baseline and the target upstream tag.
 
+## Controlled Login Profile Research
+
+Maintainers can use `scripts/research_login_matrix.py` to compare login outcomes for a reused device profile and a fresh profile. This is an opt-in diagnostic experiment for accounts and networks you control, not a CI test. Repeated login attempts can trigger Instagram checkpoints or temporary restrictions, so start with one trial and inspect the result before increasing the count.
+
+```bash
+IG_RUN_LOGIN_MATRIX=1 \
+TEST_ACCOUNTS_URL="https://your-controlled-pool.example/accounts" \
+python scripts/research_login_matrix.py --mode both --count 1
+```
+
+The account-pool URL must use HTTPS with normal certificate verification. Redirects are rejected. `stable` retains only device/profile fields from the stored client settings; it drops sessions and other authentication state. `fresh` creates a client without stored settings. The default `separate` pairing uses different accounts for the two conditions. `--pairing crossover` uses the same account for both conditions, but the first login can influence the second.
+
+Attempts run sequentially. The default cooldown is 30 seconds, the minimum is 10 seconds, and one invocation is limited to ten total login attempts. Use `--login-timeout` to bound each attempt and `--output` to select the append-only JSONL file. Non-finite numeric values are rejected.
+
+The output file is created with owner-only permissions. The script refuses symlinks and existing files accessible by group or other users, and it verifies the opened file before writing. Records include a random run ID, trial and attempt indexes, mode, status, exception class, elapsed time, a proxy-used boolean, and keyed per-run device-profile digests. They never include usernames, passwords, TOTP secrets or codes, sessions, proxy values, account-pool URLs, exception messages, or response bodies. Client and request loggers are muted during attempts so library warnings cannot print raw responses. The digest key is not saved, so device identifiers cannot be correlated across separate runs. The default `login-matrix.jsonl` output is ignored by Git.
+
+Individual login errors are experiment results and do not stop the matrix. Invalid configuration, unsafe output paths, account-pool failures, and malformed account records stop the run with a non-zero exit status. GitHub Actions never invokes this script.
+
 [pdb-docs]: https://docs.python.org/3/library/pdb.html
 [pytest-docs]: https://docs.pytest.org/en/latest/
 [ruff-docs]: https://docs.astral.sh/ruff/

scripts/research_login_matrix.py

@@ -0,0 +1,309 @@
+"""Opt-in login experiment for controlled device-profile research.
+
+The tool records pseudonymous outcome metadata only. Run it exclusively with
+accounts and networks you control.
+"""
+
+import argparse
+import asyncio
+import contextlib
+import errno
+import hashlib
+import hmac
+import json
+import logging
+import math
+import os
+import secrets
+import stat
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import httpx
+
+from aiograpi import Client
+
+MAX_ATTEMPTS = 10
+MIN_COOLDOWN_SECONDS = 10.0
+DEVICE_SETTING_KEYS = {
+    "country",
+    "country_code",
+    "device_settings",
+    "locale",
+    "timezone_offset",
+    "user_agent",
+    "uuids",
+}
+QUIET_LOGGER = logging.getLogger("aiograpi.login_matrix.quiet")
+QUIET_LOGGER.handlers.clear()
+QUIET_LOGGER.addHandler(logging.NullHandler())
+QUIET_LOGGER.propagate = False
+
+
+@dataclass(frozen=True)
+class Job:
+    trial: int
+    account_index: int
+    mode: str
+
+
+def build_accounts_url(url: str, count: int) -> str:
+    parts = urlsplit(url)
+    if parts.scheme.lower() != "https" or not parts.hostname:
+        raise ValueError("account pool URL must use HTTPS")
+    query = [(key, value) for key, value in parse_qsl(parts.query, keep_blank_values=True) if key != "count"]
+    query.append(("count", str(count)))
+    return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))
+
+
+def device_only_settings(settings: dict) -> dict:
+    return {key: value for key, value in settings.items() if key in DEVICE_SETTING_KEYS}
+
+
+def build_jobs(accounts: list[dict], modes: tuple[str, ...], pairing: str, count: int) -> list[Job]:
+    requested = count if pairing == "crossover" else count * len(modes)
+    if len(accounts) < requested:
+        raise ValueError(f"need {requested} accounts; endpoint returned {len(accounts)}")
+
+    jobs = []
+    for trial in range(count):
+        ordered_modes = modes if trial % 2 == 0 else tuple(reversed(modes))
+        for offset, mode in enumerate(ordered_modes):
+            account_index = trial if pairing == "crossover" else trial * len(modes) + offset
+            jobs.append(Job(trial=trial, account_index=account_index, mode=mode))
+    return jobs
+
+
+def digest_profile_value(key: bytes, value: object) -> str:
+    return hmac.new(key, str(value).encode(), hashlib.sha256).hexdigest()[:16]
+
+
+def profile(client: Client, digest_key: bytes) -> dict[str, str]:
+    return {
+        "uuid": digest_profile_value(digest_key, client.uuid),
+        "android_device_id": digest_profile_value(digest_key, client.android_device_id),
+        "user_agent": digest_profile_value(digest_key, client.user_agent),
+    }
+
+
+def fetch_accounts(url: str, count: int, *, getter=httpx.get) -> list[dict]:
+    try:
+        response = getter(
+            build_accounts_url(url, count),
+            headers={"User-Agent": "aiograpi-login-matrix"},
+            timeout=30,
+            follow_redirects=False,
+        )
+        response.raise_for_status()
+        payload = response.json()
+    except (httpx.HTTPError, json.JSONDecodeError, OSError, TimeoutError, ValueError) as exc:
+        raise RuntimeError(f"account pool request failed ({type(exc).__name__})") from None
+
+    accounts = payload if isinstance(payload, list) else payload.get("accounts") if isinstance(payload, dict) else None
+    if not isinstance(accounts, list):
+        raise ValueError("account pool response must contain an accounts list")
+    for account in accounts:
+        if (
+            not isinstance(account, dict)
+            or not isinstance(account.get("username"), str)
+            or not account["username"]
+            or not isinstance(account.get("password"), str)
+            or not account["password"]
+        ):
+            raise ValueError("account pool records must contain username and password")
+        for settings_key in ("client_settings", "settings"):
+            if (
+                settings_key in account
+                and account[settings_key] is not None
+                and not isinstance(account[settings_key], dict)
+            ):
+                raise ValueError("account pool settings must be objects")
+        if account.get("proxy") is not None and not isinstance(account["proxy"], str):
+            raise ValueError("account pool proxy must be a string")
+    return accounts
+
+
+async def attempt(
+    job: Job,
+    account: dict,
+    *,
+    run_id: str,
+    digest_key: bytes,
+    pairing: str,
+    login_timeout: float,
+    client_factory=Client,
+) -> dict:
+    settings = dict(account.get("client_settings") or account.get("settings") or {})
+    totp_seed = settings.pop("totp_seed", None) or account.get("totp_seed")
+    result = {
+        "run_id": run_id,
+        "trial": job.trial,
+        "mode": job.mode,
+        "pairing": pairing,
+        "proxy_used": bool(account.get("proxy")),
+        "profile_before": None,
+        "profile_after": None,
+        "status": "error",
+    }
+    started = time.monotonic()
+    client = None
+    try:
+        client = client_factory(
+            settings=device_only_settings(settings) if job.mode == "stable" else None,
+            proxy=account.get("proxy"),
+        )
+        client.logger = QUIET_LOGGER
+        client.request_logger = QUIET_LOGGER
+        client.public_request_logger = QUIET_LOGGER
+        async with client.public, client.private, client.graphql:
+            result["profile_before"] = profile(client, digest_key)
+            login_kwargs = {}
+            if totp_seed:
+                login_kwargs["verification_code"] = client.totp_generate_code(totp_seed)
+            logged_in = await asyncio.wait_for(
+                client.login(account["username"], account["password"], **login_kwargs),
+                timeout=login_timeout,
+            )
+        if logged_in:
+            result["status"] = "ok"
+        else:
+            result["error_type"] = "LoginReturnedFalse"
+    except asyncio.CancelledError:
+        raise
+    except Exception as exc:
+        result["error_type"] = type(exc).__name__
+    finally:
+        result["elapsed_ms"] = round((time.monotonic() - started) * 1000)
+        if client is not None:
+            try:
+                result["profile_after"] = profile(client, digest_key)
+            except Exception as exc:
+                result["status"] = "error"
+                result.setdefault("error_type", type(exc).__name__)
+    return result
+
+
+def selected_modes(mode: str) -> tuple[str, ...]:
+    return ("stable", "fresh") if mode == "both" else (mode,)
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--mode", choices=("stable", "fresh", "both"), default="both")
+    parser.add_argument("--count", type=int, default=1, help="number of trials")
+    parser.add_argument("--cooldown", type=float, default=30.0, help="seconds between login attempts")
+    parser.add_argument("--login-timeout", type=float, default=45.0, help="seconds allowed for each login")
+    parser.add_argument(
+        "--pairing",
+        choices=("separate", "crossover"),
+        default="separate",
+        help="use separate accounts or test both profiles on each account",
+    )
+    parser.add_argument("--output", type=Path, default=Path("login-matrix.jsonl"))
+    args = parser.parse_args(argv)
+
+    if args.count <= 0:
+        parser.error("--count must be positive")
+    if not math.isfinite(args.cooldown) or args.cooldown < MIN_COOLDOWN_SECONDS:
+        parser.error(f"--cooldown must be at least {MIN_COOLDOWN_SECONDS:g} seconds")
+    if not math.isfinite(args.login_timeout) or args.login_timeout <= 0:
+        parser.error("--login-timeout must be positive")
+    args.attempts = args.count * len(selected_modes(args.mode))
+    if args.attempts > MAX_ATTEMPTS:
+        parser.error(f"the matrix is limited to {MAX_ATTEMPTS} login attempts")
+    return args
+
+
+def require_opt_in(environ: dict[str, str]) -> str:
+    if environ.get("IG_RUN_LOGIN_MATRIX") != "1":
+        raise RuntimeError("set IG_RUN_LOGIN_MATRIX=1 to run login experiments")
+    url = environ.get("TEST_ACCOUNTS_URL", "").strip()
+    if not url:
+        raise RuntimeError("TEST_ACCOUNTS_URL is required")
+    return url
+
+
+@contextlib.contextmanager
+def secure_output(path: Path):
+    try:
+        current = os.lstat(path)
+    except FileNotFoundError:
+        current = None
+    if current is not None:
+        if stat.S_ISLNK(current.st_mode):
+            raise PermissionError("output path must not be a symlink")
+        if not stat.S_ISREG(current.st_mode):
+            raise PermissionError("output path must be a regular file")
+
+    flags = os.O_APPEND | os.O_WRONLY | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags | os.O_CREAT | os.O_EXCL, 0o600)
+    except FileExistsError:
+        try:
+            descriptor = os.open(path, flags)
+        except OSError as exc:
+            if exc.errno == errno.ELOOP:
+                raise PermissionError("output path must not be a symlink") from None
+            raise
+    try:
+        opened = os.fstat(descriptor)
+        try:
+            current = os.lstat(path)
+        except FileNotFoundError:
+            raise PermissionError("output path changed while it was opened") from None
+        if not stat.S_ISREG(opened.st_mode) or not stat.S_ISREG(current.st_mode):
+            raise PermissionError("output path must be a regular file")
+        if not os.path.samestat(opened, current):
+            raise PermissionError("output path changed while it was opened")
+        mode = stat.S_IMODE(opened.st_mode)
+        if mode & 0o077:
+            raise PermissionError("output file permissions must be owner-only")
+        output = os.fdopen(descriptor, "a", encoding="utf-8")
+        descriptor = -1
+        with output:
+            yield output
+    finally:
+        if descriptor >= 0:
+            os.close(descriptor)
+
+
+async def async_main(argv: list[str] | None = None, environ: dict[str, str] | None = None) -> int:
+    args = parse_args(argv)
+    account_pool_url = require_opt_in(os.environ if environ is None else environ)
+    modes = selected_modes(args.mode)
+    requested = args.count if args.pairing == "crossover" else args.count * len(modes)
+    accounts = await asyncio.to_thread(fetch_accounts, account_pool_url, requested)
+    jobs = build_jobs(accounts, modes, args.pairing, args.count)
+    digest_key = secrets.token_bytes(32)
+    run_id = uuid.uuid4().hex
+
+    with secure_output(args.output) as output:
+        for index, job in enumerate(jobs):
+            record = await attempt(
+                job,
+                accounts[job.account_index],
+                run_id=run_id,
+                digest_key=digest_key,
+                pairing=args.pairing,
+                login_timeout=args.login_timeout,
+            )
+            record["attempt"] = index
+            output.write(json.dumps(record, sort_keys=True) + "\n")
+            output.flush()
+            if index + 1 < len(jobs):
+                await asyncio.sleep(args.cooldown)
+    return 0
+
+
+def main(argv: list[str] | None = None, environ: dict[str, str] | None = None) -> int:
+    try:
+        return asyncio.run(async_main(argv, environ))
+    except (OSError, RuntimeError, ValueError) as exc:
+        raise SystemExit(str(exc)) from None
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())