From 1a4bd8d67275e367d634db4f10130d7e0a8f3cc1 Mon Sep 17 00:00:00 2001 From: protokey Date: Tue, 13 Oct 2026 21:48:48 +0400 Subject: [PATCH] =?UTF-8?q?feature:=20API-=D0=BA=D0=BB=D1=8E=D1=87=D0=B8?= =?UTF-8?q?=20=D0=BA=D0=BB=D0=B8=D0=B5=D0=BD=D1=82=D0=BE=D0=B2=20=D0=B8=20?= =?UTF-8?q?=D0=B6=D1=83=D1=80=D0=BD=D0=B0=D0=BB=20=D0=B8=D1=81=D0=BF=D0=BE?= =?UTF-8?q?=D0=BB=D1=8C=D0=B7=D0=BE=D0=B2=D0=B0=D0=BD=D0=B8=D1=8F=20=D0=B2?= =?UTF-8?q?=20Postgres=20(#4,=20#17)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Postgres в compose; пакет db (SQLAlchemy async): clients, windows, requests. - Все эндпоинты, кроме /health, — только с ключом в X-API-Key, иначе 401; без выпущенных ключей доступа нет. В базе — только sha256 ключа. - Задача и окно сессии привязаны к клиенту: чужой ключ их не видит. - Воркер пишет журнал: окно (открыто, последнее обращение, закрыто) и задачу (в очереди, начало, конец, итог, данные без секретов). Сбой записи не роняет задачу. - python -m db.cli create|list|revoke|stats; make key-create/key-list/key-revoke/stats. - smoke и loadtest с ключом. Co-Authored-By: Claude Opus 5.5 --- Dockerfile.api | 1 + Makefile | 31 ++++-- README.md | 16 ++- docker-compose.prod.yml | 20 ++++ docker-compose.yml | 20 ++++ docs/API.md | 8 +- docs/Deployment.md | 34 ++++++ docs/Usage.md | 3 +- dot_env | 1 + env.example | 3 +- requirements-api.txt | 3 + requirements-dev.txt | 1 + requirements.txt | 4 +- src/api/logs.py | 9 +- src/db/__init__.py | 6 ++ src/db/cli.py | 67 ++++++++++++ src/db/clients.py | 48 +++++++++ src/db/engine.py | 30 ++++++ src/db/journal.py | 69 ++++++++++++ src/db/models.py | 69 ++++++++++++ src/db/stats.py | 77 ++++++++++++++ src/main.py | 65 +++++++++--- src/tasks/broker.py | 5 +- src/tasks/store.py | 15 ++- src/worker/actors.py | 14 ++- src/worker/windows.py | 51 ++++++--- tests/harness.py | 27 ++++- tests/loadtest.py | 8 +- tests/smoke.py | 21 ++-- tests/unit/test_actors.py | 2 + tests/unit/test_auth_journal.py | 183 ++++++++++++++++++++++++++++++++ tests/unit/test_store.py | 4 +- tests/unit/test_v1.py | 8 +- tests/unit/test_v2.py | 15 +-- 34 files changed, 857 insertions(+), 81 deletions(-) create mode 100644 src/db/__init__.py create mode 100644 src/db/cli.py create mode 100644 src/db/clients.py create mode 100644 src/db/engine.py create mode 100644 src/db/journal.py create mode 100644 src/db/models.py create mode 100644 src/db/stats.py create mode 100644 tests/unit/test_auth_journal.py diff --git a/Dockerfile.api b/Dockerfile.api index 749a2cf..ca7658b 100644 --- a/Dockerfile.api +++ b/Dockerfile.api @@ -14,6 +14,7 @@ RUN pip install -r requirements-api.txt \ COPY --chown=app:app src/main.py ./src/ COPY --chown=app:app src/tasks/ ./src/tasks/ +COPY --chown=app:app src/db/ ./src/db/ COPY --chown=app:app src/api/__init__.py src/api/schemas.py ./src/api/ USER app diff --git a/Makefile b/Makefile index 4280b92..1fcb6d4 100644 --- a/Makefile +++ b/Makefile @@ -5,16 +5,17 @@ WHEELS_DIR := wheels REQUIREMENTS := requirements.txt ver ?= latest -MAKECMDGOALS_TARGETS := install freeze wheels sync sync-offline clean help build release sync-dev test test-browser up down logs smoke +MAKECMDGOALS_TARGETS := install freeze wheels sync sync-offline clean help build release sync-dev test test-browser up down logs smoke key-create key-list key-revoke stats PKG := $(filter-out $(MAKECMDGOALS_TARGETS),$(MAKECMDGOALS)) -.PHONY: install freeze wheels sync sync-offline clean help venv build release sync-dev test test-browser up down logs smoke +.PHONY: install freeze wheels sync sync-offline clean help venv build release sync-dev test test-browser up down logs smoke key-create key-list key-revoke stats # образы сервисов: api — без Chrome, worker — Chrome + Xvfb; rabbitmq и redis — сторонние, кладутся в релиз API_IMAGE := robotex-api WORKER_IMAGE := robotex-worker -DEPS_IMAGES := rabbitmq:4.1-alpine redis:8.2-alpine +DEPS_IMAGES := rabbitmq:4.1-alpine redis:8.2-alpine postgres:17-alpine API_URL ?= http://localhost:8000 +API_KEY ?= help: @echo "make install [ ...] - установить пакет(ы) в $(VENV), обновить $(REQUIREMENTS), собрать wheel в $(WHEELS_DIR)/" @@ -28,7 +29,11 @@ help: @echo "make up - собрать и запустить локально: api, worker, rabbitmq, redis" @echo "make down - остановить локальный стенд" @echo "make logs - логи api и worker" - @echo "make smoke [API_URL=...] - проверить запущенный стенд: /health, /v1/solve, /v2/tasks, удержание окна" + @echo "make smoke API_KEY=... [API_URL=...] - проверить запущенный стенд: ключ, /v1/solve, /v2/tasks, удержание окна" + @echo "make key-create name=<клиент> - выпустить API-ключ клиенту (в запущенном compose)" + @echo "make key-list - клиенты и их ключи (без самих ключей)" + @echo "make key-revoke id= - отозвать ключ клиента" + @echo "make stats [days=7] - статистика: запросы, их время, время работы браузера по клиентам" @echo "make sync-dev - установить зависимости для тестов (requirements-dev.txt)" @echo "make test - юнит-тесты без браузера" @echo "make test-browser [ver=latest] - собрать образ и прогнать сквозные тесты с Chrome в Docker (нужна сеть)" @@ -76,7 +81,20 @@ logs: docker compose logs -f api worker smoke: - $(PYTHON) tests/smoke.py $(API_URL) + $(PYTHON) tests/smoke.py $(API_URL) $(API_KEY) + +key-create: + docker compose exec api python -m db.cli create "$(name)" + +key-list: + docker compose exec api python -m db.cli list + +key-revoke: + docker compose exec api python -m db.cli revoke $(id) + +days ?= 7 +stats: + docker compose exec api python -m db.cli stats --days $(days) sync-dev: venv $(PIP) install -r requirements-dev.txt @@ -119,7 +137,8 @@ release: build 'docker compose up -d' \ '```' \ '' \ - 'Настройки (порт, слоты, CPU и память воркера, пароль RabbitMQ) — в .env.' > $(RELEASE_DIR)/README.md + 'Настройки (порт, слоты, CPU и память воркера, пароли RabbitMQ и Postgres) — в .env.' \ + 'Ключ клиенту: docker compose exec api python -m db.cli create <имя>' > $(RELEASE_DIR)/README.md @echo "Готово: $(RELEASE_DIR)/" # позволяет передавать имя пакета как аргумент: make install redis diff --git a/README.md b/README.md index 624da94..d26ed5b 100644 --- a/README.md +++ b/README.md @@ -9,10 +9,8 @@ behind Cloudflare-style JS challenges. Early MVP (v0), under active development. Single browser engine, no billing yet. Tasks and project wiki — [Redmine](https://redmine.sokolab.xyz/projects/robotex). -Not implemented yet, though planned for v1: - -- API-key auth (`X-API-Key` is accepted but **not checked**) — [#4](https://redmine.sokolab.xyz/issues/4) -- persisting request outcomes (success rate) — [#17](https://redmine.sokolab.xyz/issues/17) +Every request needs a client API key in `X-API-Key` (keys are issued with `make key-create`). Usage — requests, +their duration, browser time per client — is recorded in Postgres (`make stats`). ## Architecture @@ -44,9 +42,12 @@ Client → Caddy (TLS) → api (FastAPI) ──task──▶ RabbitMQ ──▶ the first request of a session; cookies passed in a request replace the profile's cookies. - **Browser identity is not configurable**: user agent and window size are the real browser's own — a spoofed UA mismatches `userAgentData` and Cloudflare blocks it. Clients take `user_agent` from the response. +- **Clients and usage**: Postgres — API keys (sha256 only), a journal of browser windows (opened, last used, + closed) and of requests (queued, started, finished, outcome, task data without secrets). A task and a window are + visible only to the client that created them; no keys — no access. - **Logging**: stdout, level from `LOG_LEVEL`; client secrets (proxy credentials, 2captcha key, `fill` values, cookie values) are masked. -- **Deployment**: Docker Compose — `api`, `worker`, `rabbitmq`, `redis`; TLS is terminated by Caddy on the host — see +- **Deployment**: Docker Compose — `api`, `worker`, `rabbitmq`, `redis`, `postgres`; TLS is terminated by Caddy on the host — see [docs/Deployment.md](docs/Deployment.md). ### Project layout @@ -54,6 +55,11 @@ Client → Caddy (TLS) → api (FastAPI) ──task──▶ RabbitMQ ──▶ ``` src/ main.py # api service: /v2/tasks, /v1/solve, /health, sessions + db/ # Postgres, shared by api and worker + models.py # clients, windows, requests + clients.py # API keys: issue, check, revoke + journal.py # worker: windows and requests journal + stats.py, cli.py # usage report; python -m db.cli create|list|revoke|stats tasks/ # shared by api and worker, no browser store.py # Redis: tasks, results, open windows, worker heartbeats broker.py # dramatiq broker (RabbitMQ), queues, messages diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 1cae2ed..16d035b 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -1,6 +1,7 @@ x-backend-env: &backend-env RABBITMQ_URL: amqp://robotex:${RABBITMQ_PASSWORD:-robotex}@rabbitmq:5672/ REDIS_URL: redis://redis:6379/0 + DATABASE_URL: postgresql+asyncpg://robotex:${POSTGRES_PASSWORD:-robotex}@postgres:5432/robotex services: api: @@ -15,6 +16,7 @@ services: depends_on: rabbitmq: {condition: service_healthy} redis: {condition: service_healthy} + postgres: {condition: service_healthy} worker: image: robotex-worker:latest @@ -37,6 +39,7 @@ services: depends_on: rabbitmq: {condition: service_healthy} redis: {condition: service_healthy} + postgres: {condition: service_healthy} rabbitmq: image: rabbitmq:4.1-alpine @@ -54,6 +57,22 @@ services: timeout: 10s retries: 10 + # клиенты и API-ключи, журнал окон и запросов + postgres: + image: postgres:17-alpine + restart: unless-stopped + environment: + POSTGRES_USER: robotex + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-robotex} + POSTGRES_DB: robotex + volumes: + - postgres:/var/lib/postgresql/data + healthcheck: + test: ["CMD", "pg_isready", "-U", "robotex", "-d", "robotex"] + interval: 5s + timeout: 5s + retries: 10 + redis: image: redis:8.2-alpine restart: unless-stopped @@ -69,3 +88,4 @@ volumes: api-data: rabbitmq: redis: + postgres: diff --git a/docker-compose.yml b/docker-compose.yml index 42049b1..1e572f8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -3,6 +3,7 @@ x-backend-env: &backend-env RABBITMQ_URL: amqp://robotex:${RABBITMQ_PASSWORD:-robotex}@rabbitmq:5672/ REDIS_URL: redis://redis:6379/0 + DATABASE_URL: postgresql+asyncpg://robotex:${POSTGRES_PASSWORD:-robotex}@postgres:5432/robotex services: api: @@ -20,6 +21,7 @@ services: depends_on: rabbitmq: {condition: service_healthy} redis: {condition: service_healthy} + postgres: {condition: service_healthy} worker: build: . @@ -43,6 +45,7 @@ services: depends_on: rabbitmq: {condition: service_healthy} redis: {condition: service_healthy} + postgres: {condition: service_healthy} rabbitmq: image: rabbitmq:4.1-alpine @@ -60,6 +63,22 @@ services: timeout: 10s retries: 10 + # клиенты и API-ключи, журнал окон и запросов + postgres: + image: postgres:17-alpine + restart: unless-stopped + environment: + POSTGRES_USER: robotex + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-robotex} + POSTGRES_DB: robotex + volumes: + - postgres:/var/lib/postgresql/data + healthcheck: + test: ["CMD", "pg_isready", "-U", "robotex", "-d", "robotex"] + interval: 5s + timeout: 5s + retries: 10 + redis: image: redis:8.2-alpine restart: unless-stopped @@ -83,3 +102,4 @@ volumes: extra: rabbitmq: redis: + postgres: diff --git a/docs/API.md b/docs/API.md index 3180560..f1fd766 100644 --- a/docs/API.md +++ b/docs/API.md @@ -14,7 +14,13 @@ HTML и скриншот. Задачи выполняет сервис worker и живых воркеров; `workers` — по каждому `worker_id`, `browser_slots`, `free_slots`, `windows` (открытые окна), `running` (выполняемые задачи). -> `X-API-Key` пока не проверяется ([#4](https://redmine.sokolab.xyz/issues/4)). +## Авторизация + +Все эндпоинты, кроме `/health`, требуют ключ клиента в заголовке `X-API-Key`. Без ключа, с неверным или отозванным +ключом — `401`. Ключи выпускает администратор (`make key-create name=<клиент>`, см. [Deployment.md](Deployment.md#клиенты-и-ключи)). + +Задача и окно сессии видны только клиенту, который их создал: с другим ключом `GET /v2/tasks/{id}` даёт `404`, а +чужое открытое окно не используется. ## Запрос diff --git a/docs/Deployment.md b/docs/Deployment.md index 792a778..a7e443b 100644 --- a/docs/Deployment.md +++ b/docs/Deployment.md @@ -23,6 +23,7 @@ API доступно на `http://localhost:8000`. | `worker` | `robotex-worker` (`Dockerfile`) | `dramatiq worker.actors`: выполняет задачи в Chrome, держит окна сессий (`browser_idle_timeout`) | | `rabbitmq` | `rabbitmq:4.1-alpine` | очереди: общая `robotex` и персональная `robotex.worker.` у каждого воркера | | `redis` | `redis:8.2-alpine` | задачи и результаты (TTL `RESULT_TTL_HOURS`), открытые окна (TTL = простой), heartbeat воркеров | +| `postgres` | `postgres:17-alpine` | клиенты и API-ключи, журнал окон браузера и запросов | Наружу открыт только порт `api`. Прокси и ключ 2captcha из запроса передаются только в сообщении очереди и в Redis не сохраняются. @@ -31,6 +32,37 @@ API доступно на `http://localhost:8000`. а задачи сессии с открытым окном api направляет в персональную очередь воркера, где это окно. Профили сессий лежат в volume воркера — воркеры на одном хосте делят его. +## Клиенты и ключи + +Без ключей API никого не пускает. Ключ показывается один раз при выпуске, в базе хранится только sha256. + +```bash +make key-create name=acme # выпустить ключ клиенту +make key-list # клиенты: начало ключа, активен или отозван, последнее обращение +make key-revoke id=3 # отозвать +``` + +На сервере без make: `docker compose exec api python -m db.cli create acme` (и `list`, `revoke `, `stats`). + +## Статистика использования + +Воркер пишет в Postgres: + +- `windows` — окно браузера: клиент, задача, открывшая окно, время открытия, последнего обращения и закрытия; +- `requests` — задача: клиент, `v1`/`v2`, время постановки в очередь, начала и конца, итог (`status`, `code`, + `duration_ms`), данные задачи без секретов (без логина и пароля прокси, ключа 2captcha, вводимых значений и значений + cookies); +- `clients` — время создания клиента и последнего обращения. + +Сводка по клиентам — число запросов, успешные, среднее и p95 время, окна и часы работы браузера, ошибки по `code`: + +```bash +make stats days=7 +``` + +Ошибка записи в журнал не роняет задачу — только предупреждение в логе воркера. Схема создаётся при старте api и +воркера. + ## Что внутри образа worker | Компонент | Зачем | @@ -67,6 +99,7 @@ docker compose exec worker sh -c 'rm -rf /app/extra/user_data_dir/*' - `WORKER_CPUS` / `WORKER_MEM_LIMIT` — потолок ресурсов воркера (по умолчанию 4 CPU / 8 ГБ, swap выключен). При переборе памяти OOM-killer убивает процесс Chrome внутри контейнера, хост не страдает. - `RABBITMQ_PASSWORD` — пароль пользователя `robotex` в RabbitMQ (по умолчанию `robotex`; порт наружу не открыт). +- `POSTGRES_PASSWORD` — пароль пользователя `robotex` в Postgres (по умолчанию `robotex`; порт наружу не открыт). RabbitMQ запускается с `consumer_timeout` 4 ч: задача может выполняться дольше 30 минут по умолчанию. - `stop_grace_period: 60s` у worker — при остановке он доделывает текущие задачи и закрывает окна. - `restart: unless-stopped`. @@ -85,6 +118,7 @@ docker compose exec worker sh -c 'rm -rf /app/extra/user_data_dir/*' | `NO_SLOTS_RETRY_AFTER` | `5` | api: значение `Retry-After` (с) в ответе `503` | | `RABBITMQ_URL` | `amqp://guest:guest@localhost:5672/` | api, worker; в compose задаётся сам | | `REDIS_URL` | `redis://localhost:6379/0` | api, worker; в compose задаётся сам | +| `DATABASE_URL` | `postgresql+asyncpg://robotex:robotex@localhost:5432/robotex` | api, worker; в compose задаётся сам | | `LOG_LEVEL` | `INFO` | `DEBUG` — полный дамп запроса/ответа (секреты клиента маскируются) | ## Нагрузка diff --git a/docs/Usage.md b/docs/Usage.md index db9e165..a2f4347 100644 --- a/docs/Usage.md +++ b/docs/Usage.md @@ -80,6 +80,7 @@ UA в ответе может измениться, когда на сервер | `challenge_failed` | антибот не пройден за отведённое время: повторить в той же сессии; при повторных неудачах — сменить прокси (и сессию) | | `action_failed` | упало действие: смотрите `error` и `actions[].result`; для отладки повторите с `"screenshot": true` | | `request_failed` | навигация не удалась: обычно прокси недоступен или сайт не отвечает | +| `401` | нет ключа в `X-API-Key`, ключ неверный или отозван | | `422` | ошибка в запросе: описание в `detail` | Успех — строго `status == "ok"`. @@ -97,7 +98,7 @@ import httpx ROBOTEX = "https://app.robotex.sokolab.xyz" PROXY = "http://user:pass@host:port" -robotex = httpx.Client(base_url=ROBOTEX, timeout=300) +robotex = httpx.Client(base_url=ROBOTEX, timeout=300, headers={"X-API-Key": "rbx_..."}) def solve(url: str) -> dict: diff --git a/dot_env b/dot_env index 8a2efdf..4f3728c 100644 --- a/dot_env +++ b/dot_env @@ -3,3 +3,4 @@ WORKER_CPUS=1 WORKER_MEM_LIMIT=1g BROWSER_SLOTS=1 RABBITMQ_PASSWORD= +POSTGRES_PASSWORD= diff --git a/env.example b/env.example index f0136d7..48ec18f 100644 --- a/env.example +++ b/env.example @@ -6,8 +6,9 @@ WORKER_CPUS=4 WORKER_MEM_LIMIT=8g # сколько окон Chrome на воркер — и выполняющих задачу, и удерживаемых (browser_idle_timeout); ≈ числу ядер BROWSER_SLOTS=4 -# пароль RabbitMQ (сеть compose наружу не открыта) +# пароли RabbitMQ и Postgres (сеть compose наружу не открыта) RABBITMQ_PASSWORD=robotex +POSTGRES_PASSWORD=robotex # сколько часов хранится результат задачи в Redis RESULT_TTL_HOURS=24 # через сколько дней без использования удаляется профиль браузера diff --git a/requirements-api.txt b/requirements-api.txt index 0426c45..0a264a6 100644 --- a/requirements-api.txt +++ b/requirements-api.txt @@ -15,3 +15,6 @@ starlette==1.6.0 typing-inspection==0.4.4 typing_extensions==4.16.0 uvicorn==0.52.4 +SQLAlchemy==2.1.4 +asyncpg==0.32.0 +greenlet==3.5.5 diff --git a/requirements-dev.txt b/requirements-dev.txt index 1d8f85a..9b7cf50 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -5,3 +5,4 @@ iniconfig==2.3.1 packaging==26.3 fakeredis==2.39.0 sortedcontainers==2.4.0 +aiosqlite==0.22.1 diff --git a/requirements.txt b/requirements.txt index de459fe..3353eca 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,6 +7,7 @@ annotated-doc==0.0.5 annotated-types==0.8.0 anyio==4.15.1 apify_fingerprint_datapoints==0.15.0 +asyncpg==0.32.0 attrs==26.1.0 blessed==1.49.0 browserforge==1.2.4 @@ -71,9 +72,10 @@ runs==1.3.0 screeninfo==0.8.1 shiboken6==6.11.2 six==1.17.0 +SQLAlchemy==2.1.4 starlette==1.6.0 -typing-inspection==0.4.4 typing_extensions==4.16.0 +typing-inspection==0.4.4 ua-parser==1.0.2 ua-parser-builtins==202606 urllib3==2.7.0 diff --git a/src/api/logs.py b/src/api/logs.py index 184fed3..0de2af2 100644 --- a/src/api/logs.py +++ b/src/api/logs.py @@ -13,7 +13,8 @@ def mask_proxy(proxy_url: str | None) -> str | None: return f"{parts.scheme}://{'***@' if parts.username else ''}{parts.hostname}:{parts.port}" -def safe_request_dump(payload: SolveRequest) -> str: +def safe_request_data(payload: SolveRequest) -> dict: + """Запрос без секретов: для лога и журнала запросов (#17).""" data = payload.model_dump(exclude_none=True) data["proxy"] = mask_proxy(payload.proxy) if payload.captcha: @@ -23,7 +24,11 @@ def safe_request_dump(payload: SolveRequest) -> str: for act in data["actions"]: if act["type"] in ("fill", "typing") and "value" in act: act["value"] = "***" - return json.dumps(data, indent=2, ensure_ascii=False) + return data + + +def safe_request_dump(payload: SolveRequest) -> str: + return json.dumps(safe_request_data(payload), indent=2, ensure_ascii=False) def safe_response_dump(result: dict) -> str: diff --git a/src/db/__init__.py b/src/db/__init__.py new file mode 100644 index 0000000..2f1408c --- /dev/null +++ b/src/db/__init__.py @@ -0,0 +1,6 @@ +""" +Postgres (#4, #17): клиенты с API-ключами и журнал работы — окна браузера и запросы. + +Общий для api и worker. Схема создаётся при старте сервиса (`init_db`). `DATABASE_URL` — +`postgresql+asyncpg://...`; в тестах — `sqlite+aiosqlite://...`. +""" diff --git a/src/db/cli.py b/src/db/cli.py new file mode 100644 index 0000000..b53a57b --- /dev/null +++ b/src/db/cli.py @@ -0,0 +1,67 @@ +""" +Клиенты и API-ключи: + + python -m db.cli create # выпустить ключ: печатается один раз + python -m db.cli list + python -m db.cli revoke + python -m db.cli stats [--days 7] # запросы, их время и время работы браузера по клиентам + +В compose — через make: make key-create name=..., make key-list, make key-revoke id=..., make stats days=... +""" +import argparse +import asyncio + +from db import clients, stats +from db.engine import connect + + +async def run(args) -> None: + db = connect() + try: + await db.init() + if args.command == "create": + client, key = await clients.create_client(db, args.name) + print(f"client {client.id} ({client.name}), API key — сохраните, больше он не показывается:\n{key}") + elif args.command == "list": + for c in await clients.list_clients(db): + state = f"revoked {c.revoked_at:%Y-%m-%d %H:%M}" if c.revoked_at else "active" + last = f"{c.last_access_at:%Y-%m-%d %H:%M}" if c.last_access_at else "never" + print(f"{c.id:>4} {c.key_prefix}… {state:<24} last access {last:<16} {c.name}") + elif args.command == "revoke": + print("revoked" if await clients.revoke(db, args.id) else "no active client with this id") + elif args.command == "stats": + print_stats(await stats.usage(db, stats.since_days(args.days)), args.days) + finally: + await db.close() + + +def _ms(value: int | None) -> str: + return "—" if value is None else f"{value / 1000:.1f}s" + + +def print_stats(rows: list, days: float) -> None: + print(f"за {days:g} дн.") + print(f"{'клиент':<24} {'запросов':>8} {'ok':>6} {'ср.':>7} {'p95':>7} {'окон':>5} {'браузер, ч':>10} ошибки") + for u in rows: + errors = ", ".join(f"{k} {v}" for k, v in sorted(u.errors.items())) or "—" + name = f"{u.client_id or '-'} {u.name}"[:24] + print( + f"{name:<24} {u.requests:>8} {u.ok:>6} {_ms(u.avg_ms):>7} {_ms(u.p95_ms):>7} " + f"{u.windows:>5} {u.browser_seconds / 3600:>10.2f} {errors}" + ) + if not rows: + print("запросов не было") + + +def main() -> None: + parser = argparse.ArgumentParser(prog="python -m db.cli", description="Клиенты Robotex и их API-ключи") + sub = parser.add_subparsers(dest="command", required=True) + sub.add_parser("create", help="выпустить ключ новому клиенту").add_argument("name") + sub.add_parser("list", help="список клиентов") + sub.add_parser("revoke", help="отозвать ключ").add_argument("id", type=int) + sub.add_parser("stats", help="статистика использования по клиентам").add_argument("--days", type=float, default=7) + asyncio.run(run(parser.parse_args())) + + +if __name__ == "__main__": + main() diff --git a/src/db/clients.py b/src/db/clients.py new file mode 100644 index 0000000..614f3c6 --- /dev/null +++ b/src/db/clients.py @@ -0,0 +1,48 @@ +"""Клиенты и API-ключи (#4). Ключ показывается один раз при выпуске, в базе — только sha256.""" +import hashlib +import secrets + +from sqlalchemy import select, update + +from db.engine import Database +from db.models import Client, utcnow + +KEY_PREFIX = "rbx_" + + +def hash_key(key: str) -> str: + return hashlib.sha256(key.encode()).hexdigest() + + +async def create_client(db: Database, name: str) -> tuple[Client, str]: + key = KEY_PREFIX + secrets.token_urlsafe(32) + client = Client(name=name, key_hash=hash_key(key), key_prefix=key[:12]) + async with db.session() as session, session.begin(): + session.add(client) + return client, key + + +async def authenticate(db: Database, key: str | None) -> Client | None: + """Клиент по ключу, если ключ не отозван; отмечает время последнего обращения.""" + if not key: + return None + async with db.session() as session, session.begin(): + client = await session.scalar( + select(Client).where(Client.key_hash == hash_key(key), Client.revoked_at.is_(None)) + ) + if client: + client.last_access_at = utcnow() + return client + + +async def revoke(db: Database, client_id: int) -> bool: + async with db.session() as session, session.begin(): + result = await session.execute( + update(Client).where(Client.id == client_id, Client.revoked_at.is_(None)).values(revoked_at=utcnow()) + ) + return result.rowcount > 0 + + +async def list_clients(db: Database) -> list[Client]: + async with db.session() as session: + return list(await session.scalars(select(Client).order_by(Client.id))) diff --git a/src/db/engine.py b/src/db/engine.py new file mode 100644 index 0000000..85027a6 --- /dev/null +++ b/src/db/engine.py @@ -0,0 +1,30 @@ +import os + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine + +from db.models import Base + +DEFAULT_URL = "postgresql+asyncpg://robotex:robotex@localhost:5432/robotex" +# api и worker создают схему при старте одновременно — в Postgres через advisory lock +_SCHEMA_LOCK = 780078 + + +class Database: + def __init__(self, engine: AsyncEngine): + self.engine = engine + self.session = async_sessionmaker(engine, expire_on_commit=False) + + async def init(self) -> None: + """Создать недостающие таблицы.""" + async with self.engine.begin() as conn: + if conn.dialect.name == "postgresql": + await conn.execute(text(f"SELECT pg_advisory_xact_lock({_SCHEMA_LOCK})")) + await conn.run_sync(Base.metadata.create_all) + + async def close(self) -> None: + await self.engine.dispose() + + +def connect(url: str | None = None) -> Database: + return Database(create_async_engine(url or os.getenv("DATABASE_URL", DEFAULT_URL), pool_pre_ping=True)) diff --git a/src/db/journal.py b/src/db/journal.py new file mode 100644 index 0000000..2c58d30 --- /dev/null +++ b/src/db/journal.py @@ -0,0 +1,69 @@ +""" +Журнал воркера (#17): окна браузера и задачи. Ошибка записи не роняет задачу — только предупреждение в лог. +""" +import logging +from datetime import datetime, timezone + +from sqlalchemy import update + +from db.engine import Database +from db.models import Request, Window, utcnow + +log = logging.getLogger(__name__) + + +class Journal: + def __init__(self, db: Database | None): + self.db = db + + async def _write(self, what: str, fn): + if self.db is None: + return None + try: + async with self.db.session() as session, session.begin(): + return await fn(session) + except Exception: + log.warning("journal: %s failed", what, exc_info=True) + return None + + async def window_opened(self, client_id: int | None, session_id: str, task_id: str, worker_id: str) -> int | None: + async def write(session): + window = Window(client_id=client_id, session_id=session_id, task_id=task_id, worker_id=worker_id) + session.add(window) + await session.flush() + return window.id + return await self._write("window opened", write) + + async def window_used(self, window_id: int | None) -> None: + if window_id is not None: + await self._write("window used", lambda s: s.execute( + update(Window).where(Window.id == window_id).values(last_access_at=utcnow()) + )) + + async def window_closed(self, window_id: int | None) -> None: + if window_id is not None: + await self._write("window closed", lambda s: s.execute( + update(Window).where(Window.id == window_id).values(closed_at=utcnow()) + )) + + async def request_started( + self, task_id: str, client_id: int | None, session_id: str, api: str, + created_at: float | None, task_data: dict, + ) -> None: + created = datetime.fromtimestamp(created_at, timezone.utc) if created_at else None + + async def write(session): + session.add(Request( + task_id=task_id, client_id=client_id, session_id=session_id, api=api, + created_at=created, task_data=task_data, + )) + await self._write("request started", write) + + async def request_finished(self, task_id: str, window_id: int | None, result: dict | None) -> None: + """result — ответ задачи; None — задачу не удалось выполнить (status failed).""" + values = {"finished_at": utcnow(), "window_id": window_id, "status": "failed"} + if result is not None: + values.update(status=result.get("status"), code=result.get("code"), duration_ms=result.get("duration_ms")) + await self._write("request finished", lambda s: s.execute( + update(Request).where(Request.task_id == task_id).values(**values) + )) diff --git a/src/db/models.py b/src/db/models.py new file mode 100644 index 0000000..8145eee --- /dev/null +++ b/src/db/models.py @@ -0,0 +1,69 @@ +""" +Таблицы: + +* `clients` — клиенты API («Users» в #17): ключ (только sha256), время создания и последнего обращения. +* `windows` — окна браузера: когда открыто и закрыто, последнее обращение, клиент, задача, открывшая окно. + Из них считается время, которое браузер был активен. +* `requests` — задачи: создание (постановка в очередь), начало и конец выполнения, итог, данные задачи без + секретов (прокси, ключ 2captcha, вводимые значения и значения cookies не пишутся). +""" +from datetime import datetime, timezone + +from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class Base(DeclarativeBase): + type_annotation_map = {datetime: DateTime(timezone=True)} + + +class Client(Base): + __tablename__ = "clients" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + name: Mapped[str] = mapped_column(String(200)) + # sha256 ключа: сам ключ показывается один раз при выпуске и нигде не хранится + key_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True) + # начало ключа — чтобы отличать ключи в списке + key_prefix: Mapped[str] = mapped_column(String(16)) + created_at: Mapped[datetime] = mapped_column(default=utcnow) + last_access_at: Mapped[datetime | None] + revoked_at: Mapped[datetime | None] + + +class Window(Base): + __tablename__ = "windows" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + client_id: Mapped[int | None] = mapped_column(ForeignKey("clients.id"), index=True) + session_id: Mapped[str] = mapped_column(String(36), index=True) + # задача, открывшая окно + task_id: Mapped[str] = mapped_column(String(36)) + worker_id: Mapped[str] = mapped_column(String(100)) + started_at: Mapped[datetime] = mapped_column(default=utcnow) + last_access_at: Mapped[datetime] = mapped_column(default=utcnow) + closed_at: Mapped[datetime | None] + + +class Request(Base): + __tablename__ = "requests" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + task_id: Mapped[str] = mapped_column(String(36), unique=True) + client_id: Mapped[int | None] = mapped_column(ForeignKey("clients.id"), index=True) + session_id: Mapped[str] = mapped_column(String(36), index=True) + window_id: Mapped[int | None] = mapped_column(ForeignKey("windows.id")) + # api: /v1/solve или /v2/tasks + api: Mapped[str] = mapped_column(String(8)) + created_at: Mapped[datetime | None] + started_at: Mapped[datetime] = mapped_column(default=utcnow, index=True) + finished_at: Mapped[datetime | None] + duration_ms: Mapped[int | None] + # ok / err / failed; code — как в ответе (challenge_failed, action_failed, ...) + status: Mapped[str | None] = mapped_column(String(16)) + code: Mapped[str | None] = mapped_column(String(32)) + task_data: Mapped[dict] = mapped_column(JSON) diff --git a/src/db/stats.py b/src/db/stats.py new file mode 100644 index 0000000..916c239 --- /dev/null +++ b/src/db/stats.py @@ -0,0 +1,77 @@ +"""Статистика использования API (#17): число запросов, их время и время, которое был открыт браузер.""" +from dataclasses import dataclass, field +from datetime import datetime, timedelta + +from sqlalchemy import select + +from db.engine import Database +from db.models import Client, Request, Window, utcnow + + +@dataclass +class ClientUsage: + client_id: int | None + name: str + requests: int = 0 + ok: int = 0 + # по code: challenge_failed, action_failed, request_failed; failed — задачу не выполнили + errors: dict[str, int] = field(default_factory=dict) + durations_ms: list[int] = field(default_factory=list) + windows: int = 0 + browser_seconds: float = 0.0 + + @property + def avg_ms(self) -> int | None: + return int(sum(self.durations_ms) / len(self.durations_ms)) if self.durations_ms else None + + @property + def p95_ms(self) -> int | None: + if not self.durations_ms: + return None + ordered = sorted(self.durations_ms) + return ordered[min(len(ordered) - 1, int(len(ordered) * 0.95))] + + +def _aware(dt: datetime) -> datetime: + # sqlite отдаёт время без зоны + return dt if dt.tzinfo else dt.replace(tzinfo=utcnow().tzinfo) + + +async def usage(db: Database, since: datetime) -> list[ClientUsage]: + """По клиентам за период: запросы, начатые после since, и время окон в пределах периода.""" + now = utcnow() + async with db.session() as session: + names = {c.id: c.name for c in await session.scalars(select(Client))} + requests = list(await session.scalars(select(Request).where(Request.started_at >= since))) + windows = list(await session.scalars( + select(Window).where((Window.closed_at.is_(None)) | (Window.closed_at >= since)) + )) + + result: dict[int | None, ClientUsage] = {} + + def entry(client_id): + return result.setdefault(client_id, ClientUsage(client_id, names.get(client_id, "—"))) + + for r in requests: + u = entry(r.client_id) + u.requests += 1 + if r.status == "ok": + u.ok += 1 + elif r.finished_at is not None: + key = r.code or r.status or "unknown" + u.errors[key] = u.errors.get(key, 0) + 1 + if r.duration_ms is not None: + u.durations_ms.append(r.duration_ms) + + for w in windows: + u = entry(w.client_id) + u.windows += 1 + start = max(_aware(w.started_at), since) + end = _aware(w.closed_at) if w.closed_at else now + u.browser_seconds += max(0.0, (end - start).total_seconds()) + + return sorted(result.values(), key=lambda u: -u.requests) + + +def since_days(days: float) -> datetime: + return utcnow() - timedelta(days=days) diff --git a/src/main.py b/src/main.py index 8ea69f1..0c44bd2 100644 --- a/src/main.py +++ b/src/main.py @@ -4,16 +4,21 @@ * POST /v2/tasks, GET /v2/tasks/{id}, DELETE /v2/sessions/current — асинхронный API (#80, #82). * POST /v1/solve — прежний синхронный контракт поверх той же очереди (#84). + +Все эндпоинты, кроме /health, — только с ключом клиента в X-API-Key (#4); ключи — в Postgres (db.cli). +Задача и окно сессии видны только клиенту, который их создал. """ import asyncio import logging import os import time import uuid +from contextlib import asynccontextmanager from typing import Annotated -from fastapi import Cookie, FastAPI, HTTPException, Response +from fastapi import Cookie, Depends, FastAPI, HTTPException, Response, Security from fastapi.responses import JSONResponse +from fastapi.security import APIKeyHeader from starlette.concurrency import run_in_threadpool logging.basicConfig( @@ -22,6 +27,9 @@ logging.basicConfig( ) from api.schemas import SolveRequest, SolveResponse, TaskCreated, TaskRequest, TaskResponse # noqa: E402 +from db import clients # noqa: E402 +from db.engine import connect as connect_db # noqa: E402 +from db.models import Client # noqa: E402 from tasks import broker as queue # noqa: E402 from tasks.sessions import new_session_id, parse_session_id # noqa: E402 from tasks.store import Status, TaskStore, connect # noqa: E402 @@ -34,8 +42,28 @@ V1_QUEUE_WAIT = float(os.getenv("V1_QUEUE_WAIT", "30")) POLL_INTERVAL = 0.2 store = TaskStore(connect()) +database = connect_db() -app = FastAPI() + +@asynccontextmanager +async def lifespan(app: FastAPI): + await database.init() + yield + await database.close() + + +app = FastAPI(lifespan=lifespan) +api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) + + +async def current_client(key: Annotated[str | None, Security(api_key_header)]) -> Client: + client = await clients.authenticate(database, key) + if client is None: + raise HTTPException(401, "invalid or missing X-API-Key") + return client + + +ClientDep = Annotated[Client, Depends(current_client)] NO_SLOTS = {"status": "err", "code": "no_slots", "error": "No slots available"} @@ -59,14 +87,15 @@ async def dispatch(message) -> None: await run_in_threadpool(queue.broker.enqueue, message) -async def submit(payload: TaskRequest, session_id: str, window: dict | None) -> str: +async def submit(payload: TaskRequest, session_id: str, window: dict | None, client: Client, api: str) -> str: """Зарегистрировать задачу и поставить её в очередь: в персональную очередь воркера, где открыто окно сессии (#83), иначе — в общую.""" task_id = str(uuid.uuid4()) - await store.create(task_id, session_id, payload.url) + await store.create(task_id, session_id, payload.url, client.id) target = queue.personal_queue(window["worker_id"]) if window else queue.COMMON_QUEUE await dispatch(queue.task_message( task_id, session_id, payload.model_dump(mode="json"), payload.browser_idle_timeout, target, + meta={"client_id": client.id, "api": api}, )) logger.info("task %s [%s] -> %s: %s", task_id, session_id, target, payload.url or "") return task_id @@ -76,6 +105,12 @@ async def free_slots() -> int: return sum(w["free_slots"] for w in await store.workers()) +async def client_window(session_id: str, client: Client) -> dict | None: + """Открытое окно сессии, если оно этого клиента. Чужое окно — как отсутствующее.""" + window = await store.live_window(session_id) + return window if window and window.get("client_id") == str(client.id) else None + + @app.get("/") @app.get("/health") async def health(): @@ -92,32 +127,32 @@ async def health(): async def create_task( payload: TaskRequest, response: Response, + client: ClientDep, session_id: Annotated[str | None, Cookie()] = None, ): - # TODO: добавить поддержку api ключей (#4) session_id = use_session(response, session_id) - window = await store.live_window(session_id) + window = await client_window(session_id, client) if payload.url is None and window is None: raise HTTPException(422, "url is required: the session has no open browser window") - task_id = await submit(payload, session_id, window) + task_id = await submit(payload, session_id, window, client, "v2") return {"task_id": task_id, "status": Status.QUEUED, "url": f"/v2/tasks/{task_id}"} @app.get("/v2/tasks/{task_id}", response_model=TaskResponse, response_model_exclude_none=True) -async def get_task(task_id: str, session_id: Annotated[str | None, Cookie()] = None): - # задача видна только своей сессии; чужая и несуществующая неразличимы +async def get_task(task_id: str, client: ClientDep, session_id: Annotated[str | None, Cookie()] = None): + # задача видна только своему клиенту и своей сессии; чужая и несуществующая неразличимы task = await store.get(task_id) if parse_session_id(task_id) else None - if not task or task["session_id"] != parse_session_id(session_id): + if not task or task["session_id"] != parse_session_id(session_id) or task.get("client_id") != str(client.id): raise HTTPException(404, "task not found") return {"task_id": task_id, "status": task["status"], "error": task.get("error"), "result": task.get("result")} @app.delete("/v2/sessions/current") -async def close_session(session_id: Annotated[str | None, Cookie()] = None): +async def close_session(client: ClientDep, session_id: Annotated[str | None, Cookie()] = None): """Закрыть окно сессии досрочно и освободить слот.""" session_id = parse_session_id(session_id) window = await store.get_session(session_id) if session_id else None - if not window: + if not window or window.get("client_id") != str(client.id): return {"closed": False} await store.delete_session(session_id) if await store.worker_alive(window["worker_id"]): @@ -129,19 +164,19 @@ async def close_session(session_id: Annotated[str | None, Cookie()] = None): async def solve( payload: SolveRequest, response: Response, + client: ClientDep, session_id: Annotated[str | None, Cookie()] = None, ): """Синхронный контракт v1: задача ставится в ту же очередь, ответ — когда воркер её выполнит. Если свободных слотов нет или воркер не взял задачу за V1_QUEUE_WAIT — 503, как и раньше.""" - # TODO: добавить поддержку api ключей (#4) started = time.monotonic() session_id = use_session(response, session_id) - window = await store.live_window(session_id) + window = await client_window(session_id, client) if window is None and await free_slots() <= 0: logger.warning("no free browser slots, rejecting %s", payload.url) return no_slots_response() - task_id = await submit(TaskRequest.model_validate(payload.model_dump()), session_id, window) + task_id = await submit(TaskRequest.model_validate(payload.model_dump()), session_id, window, client, "v1") while True: task = await store.get(task_id) if task is None: diff --git a/src/tasks/broker.py b/src/tasks/broker.py index e584878..734444d 100644 --- a/src/tasks/broker.py +++ b/src/tasks/broker.py @@ -39,12 +39,13 @@ dramatiq.set_broker(broker) def task_message( task_id: str, session_id: str, payload: dict, idle_timeout: int, queue: str = COMMON_QUEUE, + meta: dict | None = None, ) -> dramatiq.Message: """payload — SolveRequest/TaskRequest.model_dump(): в нём прокси и ключ 2captcha, поэтому - он живёт только в сообщении, а не в Redis.""" + он живёт только в сообщении, а не в Redis. meta — для журнала: client_id, api (v1/v2).""" return dramatiq.Message( queue_name=queue, actor_name=RUN_TASK, - args=(task_id, session_id, payload, idle_timeout), kwargs={}, options={}, + args=(task_id, session_id, payload, idle_timeout, meta or {}), kwargs={}, options={}, ) diff --git a/src/tasks/store.py b/src/tasks/store.py index 4299eb2..929e493 100644 --- a/src/tasks/store.py +++ b/src/tasks/store.py @@ -1,9 +1,9 @@ """ Состояние задач, окон и воркеров в Redis (#78). Сроки жизни — TTL ключей Redis. -* `robotex:task:` (hash) — статус задачи, сессия, время, результат. Пока задача не завершена, +* `robotex:task:` (hash) — статус задачи, сессия, клиент, время, результат. Пока задача не завершена, у ключа страховочный TTL `PENDING_TTL`; при завершении — `RESULT_TTL` (`RESULT_TTL_HOURS`). -* `robotex:session:` (hash) — где открыто окно сессии: `worker_id` и `idle_timeout`. +* `robotex:session:` (hash) — где открыто окно сессии: `worker_id`, `idle_timeout`, `client_id`. Пока задача выполняется, TTL — `RUNNING_TTL`; после неё — `browser_idle_timeout`. Ключ исчез — окно закрывается (worker/windows.py). * `robotex:worker:` — heartbeat воркера с TTL `HEARTBEAT_TTL`: слоты и число окон. @@ -62,12 +62,13 @@ class TaskStore: # --- задачи --- - async def create(self, task_id: str, session_id: str, url: str | None) -> None: + async def create(self, task_id: str, session_id: str, url: str | None, client_id: int | None = None) -> None: key = _task_key(task_id) async with self.redis.pipeline(transaction=True) as pipe: pipe.hset(key, mapping={ "status": Status.QUEUED, "session_id": session_id, + "client_id": "" if client_id is None else client_id, "url": url or "", "created_at": time.time(), "attempts": 0, @@ -156,10 +157,14 @@ class TaskStore: # --- окна сессий --- - async def set_session(self, session_id: str, worker_id: str, idle_timeout: int, ttl: int) -> None: + async def set_session( + self, session_id: str, worker_id: str, idle_timeout: int, ttl: int, client_id: int | None = None, + ) -> None: key = _session_key(session_id) async with self.redis.pipeline(transaction=True) as pipe: - pipe.hset(key, mapping={"worker_id": worker_id, "idle_timeout": idle_timeout}) + pipe.hset(key, mapping={ + "worker_id": worker_id, "idle_timeout": idle_timeout, "client_id": "" if client_id is None else client_id, + }) pipe.expire(key, ttl) await pipe.execute() diff --git a/src/worker/actors.py b/src/worker/actors.py index 6b5c218..04f0d02 100644 --- a/src/worker/actors.py +++ b/src/worker/actors.py @@ -19,6 +19,8 @@ import dramatiq # noqa: E402 from dramatiq.asyncio import get_event_loop_thread # noqa: E402 from dramatiq.middleware import Middleware # noqa: E402 +from db.engine import connect as connect_db # noqa: E402 +from db.journal import Journal # noqa: E402 from tasks import store as task_store # noqa: E402 from tasks.broker import CLOSE_WINDOW, COMMON_QUEUE, RUN_TASK, broker, personal_queue # noqa: E402 from worker.windows import Outcome, WindowPool # noqa: E402 @@ -65,7 +67,9 @@ async def _cleanup_loop(pool: WindowPool) -> None: async def _start() -> None: global _pool - _pool = WindowPool(task_store.TaskStore(task_store.connect()), WORKER_ID, BROWSER_SLOTS) + db = connect_db() + await db.init() + _pool = WindowPool(task_store.TaskStore(task_store.connect()), WORKER_ID, BROWSER_SLOTS, Journal(db)) await _pool.report() _background.extend([asyncio.create_task(_heartbeat_loop(_pool)), asyncio.create_task(_cleanup_loop(_pool))]) log.info("worker %s: %d browser slots, queues %s, %s", WORKER_ID, BROWSER_SLOTS, COMMON_QUEUE, personal_queue(WORKER_ID)) @@ -77,6 +81,8 @@ async def _stop() -> None: _background.clear() await _pool.close_all() await _pool.store.remove_worker(WORKER_ID) + if _pool.journal.db: + await _pool.journal.db.close() class Lifecycle(Middleware): @@ -95,10 +101,10 @@ broker.declare_queue(personal_queue(WORKER_ID)) @dramatiq.actor(actor_name=RUN_TASK, queue_name=COMMON_QUEUE, max_retries=0, time_limit=TASK_TIME_LIMIT_MS) -def run_task(task_id: str, session_id: str, payload: dict, idle_timeout: int) -> None: - if _run(_pool.handle(task_id, session_id, payload, idle_timeout)) == Outcome.NO_SLOT: +def run_task(task_id: str, session_id: str, payload: dict, idle_timeout: int, meta: dict | None = None) -> None: + if _run(_pool.handle(task_id, session_id, payload, idle_timeout, meta)) == Outcome.NO_SLOT: # в общую очередь: окно могло открыться на любом воркере, где есть слот - run_task.send_with_options(args=(task_id, session_id, payload, idle_timeout), delay=NO_SLOT_DELAY_MS) + run_task.send_with_options(args=(task_id, session_id, payload, idle_timeout, meta), delay=NO_SLOT_DELAY_MS) @dramatiq.actor(actor_name=CLOSE_WINDOW, queue_name=COMMON_QUEUE, max_retries=0) diff --git a/src/worker/windows.py b/src/worker/windows.py index e4af19d..ab75b11 100644 --- a/src/worker/windows.py +++ b/src/worker/windows.py @@ -29,8 +29,9 @@ from patchright.async_api import Page, ProxySettings from api import actions, profile from api.geoip import resolve_geo -from api.logs import safe_request_dump, safe_response_dump +from api.logs import safe_request_data, safe_request_dump, safe_response_dump from api.schemas import TaskRequest +from db.journal import Journal from engine import AsyncStealthySession from tasks.sessions import parse_session_id from tasks.store import RUNNING_TTL, Status, TaskStore @@ -64,6 +65,9 @@ class Window: proxy: str | None = None # окно закрыть, как только закончится текущая задача (DELETE /v2/sessions/current) closing: bool = False + client_id: int | None = None + # строка окна в журнале (#17): открыто — закрыто + journal_id: int | None = None @dataclass @@ -158,8 +162,9 @@ def build_result(payload: TaskRequest, result: dict) -> dict: class WindowPool: - def __init__(self, store: TaskStore, worker_id: str, slots: int): + def __init__(self, store: TaskStore, worker_id: str, slots: int, journal: Journal | None = None): self.store = store + self.journal = journal or Journal(None) self.worker_id = worker_id self.slots = slots self.used = 0 @@ -198,8 +203,13 @@ class WindowPool: entry = self._locks.get(session_id) return bool(entry and entry.lock.locked()) - async def handle(self, task_id: str, session_id: str, payload: dict, idle_timeout: int) -> str: - """Выполнить задачу. Результат — в Redis; возврат — что делать с сообщением очереди.""" + async def handle( + self, task_id: str, session_id: str, payload: dict, idle_timeout: int, meta: dict | None = None, + ) -> str: + """Выполнить задачу. Результат — в Redis; возврат — что делать с сообщением очереди. + meta: client_id и api (v1/v2) — для журнала.""" + meta = meta or {} + client_id = meta.get("client_id") async with self._session_lock(session_id): window = self.windows.get(session_id) new_window = window is None @@ -210,7 +220,7 @@ class WindowPool: return Outcome.NO_SLOT if task and task["status"] == Status.QUEUED else Outcome.SKIPPED # слот занимается до первого await, иначе его может перехватить другая задача self.used += 1 - window = self.windows[session_id] = Window(session_id) + window = self.windows[session_id] = Window(session_id, client_id=client_id) await self.report() if not await self.store.claim(task_id, self.worker_id, running_here=task_id in self.running): @@ -219,9 +229,16 @@ class WindowPool: return Outcome.SKIPPED self.running.add(task_id) + task = TaskRequest.model_validate(payload) + result = None + queued = await self.store.get(task_id) or {} + await self.journal.request_started( + task_id, client_id, session_id, meta.get("api", "v2"), + float(queued["created_at"]) if queued.get("created_at") else None, safe_request_data(task), + ) try: - await self.store.set_session(session_id, self.worker_id, idle_timeout, RUNNING_TTL) - result = await self._execute(window, TaskRequest.model_validate(payload)) + await self.store.set_session(session_id, self.worker_id, idle_timeout, RUNNING_TTL, client_id) + result = await self._execute(window, task, task_id) await self.store.finish(task_id, result) except Exception as e: log.exception("task %s failed", task_id) @@ -234,15 +251,16 @@ class WindowPool: raise finally: self.running.discard(task_id) + await self.journal.request_finished(task_id, window.journal_id, result) if idle_timeout and window.engine and not window.closing: - await self.store.set_session(session_id, self.worker_id, idle_timeout, idle_timeout * 60) + await self.store.set_session(session_id, self.worker_id, idle_timeout, idle_timeout * 60, client_id) else: await self._close(window) with suppress(Exception): await self.store.delete_session(session_id) return Outcome.DONE - async def _execute(self, window: Window, payload: TaskRequest) -> dict: + async def _execute(self, window: Window, payload: TaskRequest, task_id: str) -> dict: started = time.monotonic() session_id = window.session_id log.info("Incoming request [%s]: %s", session_id, payload.url or "") @@ -264,11 +282,14 @@ class WindowPool: raise NoWindow("url is required: the session has no open browser window") window.engine = await self._open(session_id, payload, nav_kwargs) window.proxy = payload.proxy - elif payload.cookies is not None: - # переданные cookies заменяют cookies профиля, как и при запуске окна - await window.engine.context.clear_cookies() - if payload.cookies: - await window.engine.context.add_cookies(payload.cookies) + window.journal_id = await self.journal.window_opened(window.client_id, session_id, task_id, self.worker_id) + else: + await self.journal.window_used(window.journal_id) + if payload.cookies is not None: + # переданные cookies заменяют cookies профиля, как и при запуске окна + await window.engine.context.clear_cookies() + if payload.cookies: + await window.engine.context.add_cookies(payload.cookies) if payload.url: raw = await window.engine.fetch( @@ -324,6 +345,8 @@ class WindowPool: async def _close_engine(self, window: Window) -> None: engine, window.engine = window.engine, None + journal_id, window.journal_id = window.journal_id, None + await self.journal.window_closed(journal_id) if engine is not None: try: await engine.close() diff --git a/tests/harness.py b/tests/harness.py index 395d707..635f990 100644 --- a/tests/harness.py +++ b/tests/harness.py @@ -1,5 +1,6 @@ """api и worker в одном процессе без RabbitMQ и Redis: Redis — fakeredis, отправка в очередь подменена доставкой сообщения прямо в WindowPool воркера (tests/unit, tests/browser).""" +import asyncio import json from contextlib import contextmanager @@ -7,6 +8,9 @@ import fakeredis from fastapi.testclient import TestClient import main +from db import clients +from db.engine import connect as connect_db +from db.journal import Journal from tasks import store as task_store from tasks.broker import RUN_TASK from worker import windows @@ -49,17 +53,34 @@ class Harness: return self.client.portal.call(fn, *args) +def make_client_key(db, name="test") -> str: + async def create(): + await db.init() + return (await clients.create_client(db, name))[1] + key = asyncio.run(create()) + asyncio.run(db.close()) # соединения sqlite привязаны к циклу asyncio.run — закрываем, приложение откроет свои + return key + + @contextmanager def run_env(monkeypatch, tmp_path, slots=2): server = fakeredis.FakeServer() store = task_store.TaskStore(fakeredis.aioredis.FakeRedis(server=server, decode_responses=True)) - pool = windows.WindowPool(store, WORKER_ID, slots=slots) + db = connect_db(f"sqlite+aiosqlite:///{tmp_path / 'robotex.db'}") + key = make_client_key(db) + pool = windows.WindowPool(store, WORKER_ID, slots=slots, journal=Journal(db)) harness = Harness(store, pool, fakeredis.FakeRedis(server=server, decode_responses=True)) - monkeypatch.setattr(windows, "USER_DATA_ROOT", tmp_path) + harness.db, harness.api_key = db, key + # профили — в отдельном каталоге: в tmp_path ещё и файл базы + profiles = tmp_path / "profiles" + profiles.mkdir() + harness.profiles = profiles + monkeypatch.setattr(windows, "USER_DATA_ROOT", profiles) monkeypatch.setattr(main, "store", store) + monkeypatch.setattr(main, "database", db) monkeypatch.setattr(main, "dispatch", harness.dispatch) harness.heartbeat() - with TestClient(main.app) as client: + with TestClient(main.app, headers={"X-API-Key": key}) as client: harness.client = client yield harness client.portal.call(pool.close_all) diff --git a/tests/loadtest.py b/tests/loadtest.py index 53e305b..2cf2abc 100644 --- a/tests/loadtest.py +++ b/tests/loadtest.py @@ -10,6 +10,7 @@ import argparse import asyncio import json +import os import statistics import subprocess import threading @@ -94,7 +95,8 @@ async def run_level(n: int, args) -> None: sampler = StatsSampler(args.container) sampler.start() t0 = time.monotonic() - async with httpx.AsyncClient(timeout=args.http_timeout, trust_env=False) as client: + headers = {"X-API-Key": args.api_key} if args.api_key else {} + async with httpx.AsyncClient(timeout=args.http_timeout, trust_env=False, headers=headers) as client: results = await asyncio.gather(*(one(client, args.api, build_payload(args)) for _ in range(n))) wall = time.monotonic() - t0 sampler.stop_event.set() @@ -120,7 +122,9 @@ async def main(): p.add_argument("--selector", default='a[href*="/project/"]') p.add_argument("--proxy", default="http://proxy:8080") p.add_argument("--captcha-type", default="cloudflare") - p.add_argument("--container", default="webroboapi-api-1") + p.add_argument("--api-key", default=os.getenv("ROBOTEX_API_KEY"), help="X-API-Key (или ROBOTEX_API_KEY)") + # Chrome работает в воркере — его docker stats и снимаем + p.add_argument("--container", default="robotex-worker-1") p.add_argument("--levels", type=int, nargs="+", default=[1, 5, 10]) p.add_argument("--pause", type=float, default=10, help="пауза между уровнями, с") p.add_argument("--http-timeout", type=float, default=300) diff --git a/tests/smoke.py b/tests/smoke.py index 341f741..f076a60 100644 --- a/tests/smoke.py +++ b/tests/smoke.py @@ -1,7 +1,9 @@ """Проверка запущенного стенда (make up): api, очередь, воркер с Chrome, удержание окна. - make smoke # http://localhost:8000 - make smoke API_URL=https://app.robotex.sokolab.xyz + make smoke API_KEY=rbx_... # http://localhost:8000 + make smoke API_KEY=rbx_... API_URL=https://app.robotex.sokolab.xyz + +Ключ: make key-create name=smoke. """ import sys import time @@ -32,8 +34,13 @@ def windows(client: httpx.Client) -> int: return sum(w["windows"] for w in client.get("/health").json()["workers"]) -def main(base_url: str) -> None: - client = httpx.Client(base_url=base_url, timeout=300) +def main(base_url: str, api_key: str) -> None: + headers = {"X-API-Key": api_key} + client = httpx.Client(base_url=base_url, timeout=300, headers=headers) + + print("auth") + resp = httpx.post(f"{base_url}/v1/solve", json={"url": URL, "actions": []}, timeout=30) + check(resp.status_code == 401, f"no key -> {resp.status_code}") print("health") health = client.get("/health").json() @@ -46,7 +53,7 @@ def main(base_url: str) -> None: check(data["actions"][0]["result"] == "Example Domain", "title = Example Domain") print("v2 window held between tasks") - v2 = httpx.Client(base_url=base_url, timeout=60) # своя сессия + v2 = httpx.Client(base_url=base_url, timeout=60, headers=headers) # своя сессия started = time.monotonic() created = v2.post("/v2/tasks", json={"url": URL, "actions": READ_TITLE, "browser_idle_timeout": 2}) check(created.status_code == 202, f"POST /v2/tasks -> {created.status_code}") @@ -73,4 +80,6 @@ def main(base_url: str) -> None: if __name__ == "__main__": - main(sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8000") + if len(sys.argv) < 3 or not sys.argv[2]: + sys.exit("usage: smoke.py (make smoke API_KEY=...)") + main(sys.argv[1], sys.argv[2]) diff --git a/tests/unit/test_actors.py b/tests/unit/test_actors.py index 9646810..70fdc29 100644 --- a/tests/unit/test_actors.py +++ b/tests/unit/test_actors.py @@ -22,7 +22,9 @@ def worker(monkeypatch, tmp_path): monkeypatch.setattr(actions, "execute", fake_execute()) FakeSession.instances, FakeSession.error = [], None + from db.engine import connect as connect_db from worker import actors + monkeypatch.setattr(actors, "connect_db", lambda: connect_db(f"sqlite+aiosqlite:///{tmp_path / 'robotex.db'}")) w = dramatiq.Worker(actors.broker, worker_timeout=100, worker_threads=2) w.start() try: diff --git a/tests/unit/test_auth_journal.py b/tests/unit/test_auth_journal.py new file mode 100644 index 0000000..435e3d8 --- /dev/null +++ b/tests/unit/test_auth_journal.py @@ -0,0 +1,183 @@ +"""API-ключи (#4) и журнал окон и запросов (#17).""" +import argparse +import asyncio +import uuid + +import pytest +from sqlalchemy import select + +from db import cli, clients +from db.engine import connect as connect_db +from db.models import Client, Request, Window + +from .conftest import FakeSession + +URL = "https://example.com" + + +def rows(env, model): + async def query(): + async with env.db.session() as session: + return list(await session.scalars(select(model).order_by(model.id))) + return env.call(query) + + +def new_key(env, name="other"): + return env.call(clients.create_client, env.db, name)[1] + + +@pytest.mark.parametrize("request_", [ + ("post", "/v1/solve", {"url": URL, "actions": []}), + ("post", "/v2/tasks", {"url": URL, "actions": []}), + ("get", f"/v2/tasks/{uuid.uuid4()}", None), + ("delete", "/v2/sessions/current", None), +]) +@pytest.mark.parametrize("key", [None, "", "rbx_wrong"]) +def test_requests_without_valid_key_rejected(env, request_, key): + method, path, body = request_ + headers = {"X-API-Key": key} if key is not None else {} + env.client.headers.pop("X-API-Key") + resp = env.client.request(method, path, json=body, headers=headers) + assert resp.status_code == 401 + assert env.messages == [] + + +def test_health_is_open(env): + env.client.headers.pop("X-API-Key") + assert env.client.get("/health").status_code == 200 + + +def test_revoked_key_rejected(env): + assert env.client.post("/v1/solve", json={"url": URL, "actions": []}).status_code == 200 + assert env.call(clients.revoke, env.db, 1) + assert env.client.post("/v1/solve", json={"url": URL, "actions": []}).status_code == 401 + + +def test_no_clients_means_no_access(env): + env.call(clients.revoke, env.db, 1) + assert [c.revoked_at is not None for c in rows(env, Client)] == [True] + assert env.client.post("/v2/tasks", json={"url": URL, "actions": []}).status_code == 401 + + +def test_last_access_tracked(env): + assert rows(env, Client)[0].last_access_at is None + env.client.post("/v2/tasks", json={"url": URL, "actions": []}) + assert rows(env, Client)[0].last_access_at is not None + + +def test_task_and_window_visible_only_to_their_client(env): + task_id = env.client.post("/v2/tasks", json={"url": URL, "actions": [], "browser_idle_timeout": 5}).json()["task_id"] + other = {"X-API-Key": new_key(env)} + + # та же cookie сессии, но другой ключ: задача и окно — как чужие + assert env.client.get(f"/v2/tasks/{task_id}", headers=other).status_code == 404 + assert env.client.post("/v2/tasks", json={"actions": []}, headers=other).status_code == 422 + assert env.client.delete("/v2/sessions/current", headers=other).json() == {"closed": False} + assert env.pool.used == 1 + + assert env.client.get(f"/v2/tasks/{task_id}").json()["status"] == "done" + assert env.client.delete("/v2/sessions/current").json() == {"closed": True} + + +def test_journal_request_and_window(env): + env.client.post("/v2/tasks", json={ + "url": URL, "proxy": "http://user:TopSecret1@host:3128", "captcha": "KEYSECRET2", + "actions": [{"type": "typing", "selector": "#p", "value": "PassSecret3"}], + }) + env.client.post("/v1/solve", json={"url": URL, "actions": []}) + + requests = rows(env, Request) + assert [(r.api, r.status, r.client_id) for r in requests] == [("v2", "ok", 1), ("v1", "ok", 1)] + for r in requests: + assert r.created_at <= r.started_at <= r.finished_at + assert r.duration_ms is not None and r.window_id is not None + data = str(requests[0].task_data) + assert requests[0].task_data["url"] == URL + for secret in ("user", "TopSecret1", "KEYSECRET2", "PassSecret3"): + assert secret not in data + + windows = rows(env, Window) + assert len(windows) == 2 + assert all(w.closed_at is not None and w.started_at <= w.closed_at for w in windows) + assert [w.task_id for w in windows] == [r.task_id for r in requests] + + +def test_journal_held_window(env): + for body in ({"url": URL, "browser_idle_timeout": 5}, {"browser_idle_timeout": 5}): + env.client.post("/v2/tasks", json={"actions": [], **body}) + window, = rows(env, Window) + assert window.closed_at is None and window.last_access_at > window.started_at + assert {r.window_id for r in rows(env, Request)} == {window.id} + + env.client.delete("/v2/sessions/current") + assert rows(env, Window)[0].closed_at is not None + + +def test_journal_failed_request(env): + FakeSession.error = RuntimeError("net::ERR_PROXY_CONNECTION_FAILED") + env.client.post("/v2/tasks", json={"url": URL, "actions": []}) + request, = rows(env, Request) + assert (request.status, request.code) == ("err", "request_failed") + + +def test_journal_failure_does_not_break_task(env, monkeypatch): + class DownDatabase: + def session(self): + raise RuntimeError("db is down") + monkeypatch.setattr(env.pool.journal, "db", DownDatabase()) + task_id = env.client.post("/v2/tasks", json={"url": URL, "actions": []}).json()["task_id"] + assert env.client.get(f"/v2/tasks/{task_id}").json()["result"]["status"] == "ok" + + +def test_cli_create_list_revoke(monkeypatch, tmp_path, capsys): + monkeypatch.setenv("DATABASE_URL", f"sqlite+aiosqlite:///{tmp_path / 'cli.db'}") + asyncio.run(cli.run(argparse.Namespace(command="create", name="acme"))) + key = capsys.readouterr().out.strip().splitlines()[-1] + assert key.startswith(clients.KEY_PREFIX) + + asyncio.run(cli.run(argparse.Namespace(command="list"))) + listing = capsys.readouterr().out + assert "acme" in listing and "active" in listing and key not in listing + + asyncio.run(cli.run(argparse.Namespace(command="revoke", id=1))) + assert capsys.readouterr().out.strip() == "revoked" + + asyncio.run(cli.run(argparse.Namespace(command="stats", days=7))) + assert "запросов не было" in capsys.readouterr().out + + async def check(): + db = connect_db() + try: + return await clients.authenticate(db, key) + finally: + await db.close() + assert asyncio.run(check()) is None + + +def test_usage_stats(env, monkeypatch): + from api import actions as api_actions + from db import stats + + from .conftest import fake_execute + + env.client.post("/v2/tasks", json={"url": URL, "actions": [], "browser_idle_timeout": 5}) + env.client.post("/v2/tasks", json={"actions": []}) # в открытом окне, после неё окно закрывается + monkeypatch.setattr(api_actions, "execute", fake_execute(exc=api_actions.ChallengeFailed("cf"))) + env.client.post("/v1/solve", json={"url": URL, "actions": []}) + + other = {"X-API-Key": new_key(env, "acme")} + env.client.cookies.clear() + monkeypatch.setattr(api_actions, "execute", fake_execute()) + env.client.post("/v1/solve", json={"url": URL, "actions": []}, headers=other) + + usage = {u.name: u for u in env.call(stats.usage, env.db, stats.since_days(1))} + test, acme = usage["test"], usage["acme"] + assert (test.requests, test.ok, test.errors, test.windows) == (3, 2, {"challenge_failed": 1}, 2) + assert (acme.requests, acme.ok, acme.windows) == (1, 1, 1) + assert test.avg_ms is not None and test.p95_ms is not None + assert test.browser_seconds > 0 + # все окна закрыты: время браузера больше не растёт + again = {u.name: u.browser_seconds for u in env.call(stats.usage, env.db, stats.since_days(1))} + assert again == {name: u.browser_seconds for name, u in usage.items()} + # период фильтрует запросы и окна + assert env.call(stats.usage, env.db, stats.since_days(-1)) == [] diff --git a/tests/unit/test_store.py b/tests/unit/test_store.py index 33e2336..6d35d64 100644 --- a/tests/unit/test_store.py +++ b/tests/unit/test_store.py @@ -67,8 +67,8 @@ async def test_finish_sets_result_and_ttl(store): async def test_live_window_requires_live_worker(store): - await store.set_session("s", "w1", 5, 300) + await store.set_session("s", "w1", 5, 300, client_id=7) assert await store.live_window("s") is None await store.heartbeat("w1", {"browser_slots": 1, "free_slots": 0}) - assert await store.live_window("s") == {"worker_id": "w1", "idle_timeout": "5"} + assert await store.live_window("s") == {"worker_id": "w1", "idle_timeout": "5", "client_id": "7"} assert await store.workers() == [{"worker_id": "w1", "browser_slots": 1, "free_slots": 0}] diff --git a/tests/unit/test_v1.py b/tests/unit/test_v1.py index c077e10..7a0d6f3 100644 --- a/tests/unit/test_v1.py +++ b/tests/unit/test_v1.py @@ -109,21 +109,21 @@ def test_invalid_request_is_422_without_task(env): assert env.messages == [] -def test_session_cookie_reused(env, tmp_path): +def test_session_cookie_reused(env): sid = str(uuid.uuid4()) env.client.cookies.set("session_id", sid) resp = post(env) assert resp.cookies["session_id"] == sid - assert FakeSession.instances[0].kwargs["user_data_dir"] == str(tmp_path / sid) + assert FakeSession.instances[0].kwargs["user_data_dir"] == str(env.profiles / sid) @pytest.mark.parametrize("bad", ["../../etc", "not-a-uuid", "/tmp/x"]) -def test_bad_session_cookie_gets_new_session(env, tmp_path, bad): +def test_bad_session_cookie_gets_new_session(env, bad): env.client.cookies.set("session_id", bad) resp = post(env) sid = resp.cookies["session_id"] assert sid != bad and uuid.UUID(sid) - assert [p.name for p in tmp_path.iterdir()] == [sid] + assert [p.name for p in env.profiles.iterdir()] == [sid] def test_user_agent_param_rejected(env): diff --git a/tests/unit/test_v2.py b/tests/unit/test_v2.py index abe782a..6f7b1c7 100644 --- a/tests/unit/test_v2.py +++ b/tests/unit/test_v2.py @@ -90,7 +90,8 @@ def test_window_held_and_reused(env): sid = env.client.cookies["session_id"] engine = FakeSession.instances[0] assert not engine.closed and env.pool.used == 1 - assert env.redis.hgetall(session_key(sid)) == {"worker_id": WORKER_ID, "idle_timeout": "5"} + window = env.redis.hgetall(session_key(sid)) + assert (window["worker_id"], window["idle_timeout"], window["client_id"]) == (WORKER_ID, "5", "1") assert 290 < env.redis.ttl(session_key(sid)) <= 300 # следующая задача сессии — в персональную очередь воркера с окном, без url — на текущей странице @@ -180,7 +181,7 @@ def test_dead_worker_window_goes_to_common_queue(env): assert env.messages[-1].queue_name == COMMON_QUEUE -def test_old_profiles_cleaned_up(env, tmp_path): +def test_old_profiles_cleaned_up(env): """#20: профили старше PROFILE_TTL_DAYS удаляются, кроме тех, где открыто окно.""" import os import time @@ -194,18 +195,18 @@ def test_old_profiles_cleaned_up(env, tmp_path): long_ago = time.time() - 8 * 86400 for sid in (held, elsewhere, old, fresh, "not-a-session"): - (tmp_path / sid).mkdir(exist_ok=True) - marker = tmp_path / sid / LAST_USED_FILE + (env.profiles / sid).mkdir(exist_ok=True) + marker = env.profiles / sid / LAST_USED_FILE marker.touch() if sid != fresh: os.utime(marker, (long_ago, long_ago)) assert env.call(env.pool.cleanup_profiles) == [old] - assert sorted(p.name for p in tmp_path.iterdir()) == sorted([held, elsewhere, fresh, "not-a-session"]) + assert sorted(p.name for p in env.profiles.iterdir()) == sorted([held, elsewhere, fresh, "not-a-session"]) -def test_profile_marked_as_used(env, tmp_path): +def test_profile_marked_as_used(env): from worker.windows import LAST_USED_FILE create(env, url="https://example.com") - assert (tmp_path / env.client.cookies["session_id"] / LAST_USED_FILE).exists() + assert (env.profiles / env.client.cookies["session_id"] / LAST_USED_FILE).exists()