Keitaro Remote Control

Async FastAPI microservice for managing multiple Keitaro tracker instances remotely.

Quick Start (Production)

cp .env.example .env
# Edit .env with your keys (JWT_SECRET_KEY, TOKEN_ENCRYPTION_KEY, MARIADB_PASSWORD)

docker compose up -d

Service runs on port 8300.

Local Development

Requires uv and Docker.

1. Start infrastructure

docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d mariadb redis

This starts MariaDB and Redis from docker-compose.yml with port overrides from docker-compose.dev.yml (MariaDB on 3306, Redis on 6379).

2. Install dependencies

uv sync

3. Configure environment

cp .env.example .env

Edit .env and set:

• JWT_SECRET_KEY — any random string (see key generation below)
• TOKEN_ENCRYPTION_KEY — Fernet key (see key generation below)
• MARIADB_PASSWORD — must match the password in docker-compose.dev.yml (default: defaultpassword)
• MARIADB_HOST=localhost
• RABBITMQ_HOST=localhost
• REDIS_URL=redis://localhost:6379/0
• CACHE_TTL_SECONDS=300 — cache entry TTL in seconds (default: 300)
• CACHE_ENABLED=true — set to false to bypass caching entirely

4. Run migrations and start the server

# Apply database migrations
uv run alembic upgrade head

# Start dev server (hot reload)
uv run fastapi dev app/main.py --host 0.0.0.0 --port 8300

5. Run tests

# All tests
uv run pytest

# Unit tests only (no running services required)
uv run pytest -m "not integration"

Stop infrastructure

docker compose -f docker-compose.yml -f docker-compose.dev.yml down

To also remove volumes (reset DB and Redis data):

docker compose -f docker-compose.yml -f docker-compose.dev.yml down -v

Generate required keys

# Generate TOKEN_ENCRYPTION_KEY (Fernet key)
python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

# Generate JWT_SECRET_KEY (any random string)
python -c "import secrets; print(secrets.token_urlsafe(32))"

Architecture

Request → main.py → api/v1/api.py → modules/{feature}/router.py → service.py → repository.py → DB

All module routers flow through api/v1/api.py. No shortcuts.

Tech Stack

Layer	Technology
Framework	FastAPI 0.136 + Uvicorn 0.46
Runtime	Python 3.14, fully async
Package Manager	uv
Database	MariaDB 12 via SQLAlchemy 2.0.49 (async) + aiomysql
Cache	Redis 8 via redis-py (async) — entity list caching with TTL
Migrations	Alembic
Task Queue	TaskIQ + RabbitMQ (delayed exchange) + Redis backend
Auth	PyJWT (HS256) — shared secret with toolbox & gambling
Encryption	Fernet (cryptography 48.0) for API key storage

Modules

Instances — /api/v1/instances/

CRUD for registered Keitaro tracker instances.

Method	Path	Description
GET	/	List all instances (API keys masked)
POST	/	Register new instance (auto-creates default key)
GET	/{id}	Get instance details
PUT	/{id}	Update instance
DELETE	/{id}	Remove instance (cascades to keys)

API Keys — /api/v1/instances/{id}/keys/

Manage multiple API keys per instance (up to 20).

Method	Path	Description
GET	/	List all keys for instance (masked)
POST	/	Add a new API key
PUT	/{key_id}	Update key label or activate
DELETE	/{key_id}	Delete a key
GET	/{key_id}/reveal	Reveal decrypted key value
POST	/check	Health check all keys for instance
POST	/{key_id}/check	Health check a single key

Campaigns — /api/v1/campaigns/

Remote campaign operations across any registered instance.

Method	Path	Description
GET	/instance/{id}	List campaigns on instance (response includes total, the full-list count before pagination)
GET	/instance/{id}/{campaign_id}	Get single campaign
GET	/instance/{id}/{campaign_id}/flows	List flows (streams) for a campaign, enriched with offer/landing names (cached)
POST	/instance/{id}	Create campaign
PUT	/instance/{id}/{campaign_id}	Update campaign
DELETE	/instance/{id}/{campaign_id}	Delete campaign
POST	/bulk	Create same campaign across multiple instances

Offers — /api/v1/offers/

Retrieve offers from any registered Keitaro instance.

Method	Path	Description
GET	/instance/{id}	List all offers on instance (response includes total, the full-list count before pagination)
GET	/instance/{id}/{offer_id}	Get single offer
GET	/instance/{id}/{offer_id}/download	Stream offer zip download (binary proxy)
PUT	/instance/{id}/{offer_id}/upload	Upload offer zip (base64-encoded archive)

Landings — /api/v1/landings/

Retrieve landing pages from any registered Keitaro instance.

Method	Path	Description
GET	/instance/{id}	List all landing pages on instance (response includes total, the full-list count before pagination)
GET	/instance/{id}/{landing_id}	Get single landing page
GET	/instance/{id}/{landing_id}/download	Stream landing page zip download (binary proxy)
PUT	/instance/{id}/{landing_id}/upload	Upload landing page zip (base64-encoded archive)

Domains — /api/v1/domains/

Retrieve domains from any registered Keitaro instance.

Method	Path	Description
GET	/instance/{id}	List all domains on instance (response includes total, the full-list count before pagination)
GET	/instance/{id}/{domain_id}	Get single domain

Traffic Sources — /api/v1/traffic-sources/

Retrieve traffic sources from any registered Keitaro instance.

Method	Path	Description
GET	/instance/{id}	List all traffic sources on instance
GET	/instance/{id}/{traffic_source_id}	Get single traffic source

Affiliate Networks — /api/v1/affiliate-networks/

Retrieve affiliate networks from any registered Keitaro instance.

Method	Path	Description
GET	/instance/{id}	List all affiliate networks on instance
GET	/instance/{id}/{network_id}	Get single affiliate network

Users — /api/v1/users/

Retrieve Keitaro tracker users from any registered instance. These are users within the Keitaro tracker, not KRC authentication users.

Method	Path	Description
GET	/instance/{id}	List all Keitaro users on instance
GET	/instance/{id}/{user_id}	Get single Keitaro user

Groups — /api/v1/groups/

Retrieve groups (campaigns, offers, landings, domains) from any registered Keitaro instance.

Method	Path	Description
GET	/instance/{id}	List groups (optional ?type= filter)
GET	/instance/{id}/{group_id}	Get single group

Reports — /api/v1/reports/

Build reports on any registered Keitaro instance.

Method	Path	Description
POST	/instance/{id}	Build a report (with range, grouping, metrics, filters, pagination)

Supports optional limit and offset fields in the request body for in-memory pagination of results. Keitaro always returns all matching rows; the backend slices them before responding. The total field in the response reflects the full (unsliced) row count.

Conversions — /api/v1/conversions/

Retrieve conversion logs from any registered Keitaro instance.

Method	Path	Description
POST	/instance/{id}	Get conversion log (with range, columns, filters, sort, pagination)

Health — /api/v1/health/

Method	Path	Description
GET	/service	Service + DB health check
POST	/ping-all	Check all registered Keitaro instances
POST	/instance/{id}/ping	Check single instance

Tasks — /api/v1/tasks/

Method	Path	Description
GET	/{task_id}/status	Poll background task status

Database Models

kr_instances — Registered Keitaro instances

• name (unique), base_url, ip_address (optional, max 45 chars), is_active, description

kr_api_keys — API keys (one-to-many with instances)

• instance_id (FK, cascade delete), encrypted_key, label (unique per instance), is_active, status (unknown/working/invalid/error), last_checked_at

API Key Security

• Keys are Fernet-encrypted in the kr_api_keys table
• Normal API responses return masked keys: "abcd****7890"
• Full decryption only via /keys/{key_id}/reveal endpoint
• Multiple keys per instance (up to 20) with labels
• Exactly one key per instance is designated as active
• Key rotation: add new key, activate it, delete old one
• Per-key health status tracking (unknown → working/invalid/error)

Docker Compose Services

Currently active:

Service	Container	Port
API	krc-api	8300
MariaDB	krc-mariadb	3306
Redis	krc-redis	6379

Commented out (planned for TaskIQ integration):

Service	Container	Port
RabbitMQ	krc-rabbitmq	5672
TaskIQ Worker	krc-taskiq-worker	—
TaskIQ Scheduler	krc-taskiq-scheduler	—
TaskIQ Admin	krc-taskiq-admin	3000

Configuration

All config via environment variables loaded through pydantic-settings in app/core/config.py.

Variable	Required	Description
API_PORT	No	Port to listen on (default: 8300)
JWT_SECRET_KEY	Yes	Shared JWT secret for auth
TOKEN_ENCRYPTION_KEY	Yes	Fernet key for API key encryption (generate with cryptography)
MARIADB_HOST	No	DB host (default: localhost)
MARIADB_PORT	No	DB port (default: 3306)
MARIADB_USER	No	DB user (default: krc_user)
MARIADB_PASSWORD	Yes	DB password
MARIADB_DATABASE	No	DB name (default: keitaro_remote_control)
RABBITMQ_HOST	No	RabbitMQ host (default: localhost)
RABBITMQ_PORT	No	RabbitMQ port (default: 5672)
RABBITMQ_USER	No	RabbitMQ user (default: guest)
RABBITMQ_PASSWORD	No	RabbitMQ password (default: guest)
REDIS_URL	No	Redis URL (default: redis://localhost:6379/0)
CACHE_TTL_SECONDS	No	Cache entry TTL in seconds, range 1–86400 (default: 300)
CACHE_ENABLED	No	Enable/disable entity caching (default: true)
ENVIRONMENT	No	development or production
SENTRY_DSN	No	Sentry DSN (production only)
LOG_LEVEL	No	Application log level (default: INFO)
LOG_DIR	No	Directory for file-based logs (default: logs)
UPLOAD_MAX_SIZE_MB	No	Max upload file size in MB (default: 50)
UPLOAD_TIMEOUT_SECONDS	No	Timeout for upload requests to Keitaro (default: 60)

Entity Caching

Redis-based cache-aside pattern for entity list endpoints (campaigns, offers, landings, domains). Reduces repeated round-trip API calls to remote Keitaro instances.

Behavior:

• On list request: check Redis → cache hit returns immediately; cache miss fetches from Keitaro, stores in Redis, returns
• On write (create/update/delete/upload): invalidates the cache for that entity type on that instance (domains has no write endpoints yet, so nothing invalidates it)
• On flows request: offer/landing names are resolved from cache (or fetched and cached) for enrichment
• Full list is always cached; pagination (offset/limit) is applied in-memory after retrieval, and the response carries total — the full-list count before slicing
• List responses are typed Pydantic models (CampaignListResponse, OfferListResponse, LandingListResponse, DomainListResponse), each shaped {instance_id, <entity>, total}
• All Redis failures degrade gracefully — the app continues serving from Keitaro directly
• Cache keys follow the pattern krc:{instance_id}:{entity_type}

Configuration:

• CACHE_ENABLED=true — global on/off switch (all operations become no-ops when false)
• CACHE_TTL_SECONDS=300 — automatic expiration safety net (1–86400)
• REDIS_URL — connection endpoint (shared with TaskIQ result backend)

KeitaroClient

Located in app/services/keitaro_client.py. Built from scratch — generic, multi-instance. Each client instance takes its own base_url + api_key, enabling operations across any registered tracker.

Supported Keitaro endpoints:

• GET /ping — health check
• GET /campaigns — list campaigns
• GET /campaigns/{id} — get campaign
• GET /campaigns/{id}/streams — list campaign flows (streams)
• POST /campaigns — create campaign
• PUT /campaigns/{id} — update campaign
• DELETE /campaigns/{id} — delete campaign
• GET /groups — list groups (with optional ?type= filter)
• GET /groups/{id} — get group
• GET /offers — list offers
• GET /offers/{id} — get offer
• PUT /offers/{id} — update offer (upload archive, configurable timeout)
• GET /offers/{id}/download — stream offer zip (binary, via _stream_request())
• GET /landing_pages — list landing pages
• GET /landing_pages/{id} — get landing page
• PUT /landing_pages/{id} — update landing page (upload archive, configurable timeout)
• GET /landing_pages/{id}/download — stream landing page zip (binary, via _stream_request())
• GET /domains — list domains
• GET /domains/{id} — get domain
• GET /traffic_sources — list traffic sources
• GET /traffic_sources/{id} — get traffic source
• GET /affiliate_networks — list affiliate networks
• GET /affiliate_networks/{id} — get affiliate network
• GET /users — list Keitaro tracker users
• GET /users/{id} — get Keitaro tracker user
• POST /conversions/log — get conversion log
• GET /reports/main — fetch reports
• POST /report/build — build report

Two request methods:

• _request() — standard JSON endpoints (parses response body)
• _stream_request() — binary streaming endpoints (returns raw httpx.Response with 30s timeout, caller responsible for closing)

Upload methods (update_offer, update_landing) use a separate configurable timeout (UPLOAD_TIMEOUT_SECONDS) to accommodate large file transfers.

Common Commands

# Run all tests
uv run pytest

# Run tests excluding integration
uv run pytest -m "not integration"

# Create a new migration
uv run alembic revision --autogenerate -m "description"

# Apply migrations
uv run alembic upgrade head

# Start dev server
uv run fastapi dev app/main.py --host 0.0.0.0 --port 8300

# Export requirements.txt (for Docker builds)
uv export --no-dev --no-hashes --frozen -o requirements.txt

Project Structure

keitaro-remote-control/
├── app/
│   ├── main.py                      # FastAPI app, lifespan, router mount
│   ├── core/
│   │   ├── config.py                # pydantic-settings
│   │   ├── logging.py               # console + file logging
│   │   └── taskiq.py                # broker, scheduler, middlewares
│   ├── api/
│   │   ├── deps.py                  # JWT auth dependency
│   │   ├── streaming.py             # Shared streaming download helpers
│   │   └── v1/
│   │       └── api.py               # Central router (mounts all modules)
│   ├── db/
│   │   ├── base.py                  # async engine, session, Base model
│   │   └── models/
│   │       ├── kr_instance.py       # Keitaro instance model
│   │       └── kr_api_key.py        # API key model (multi-key per instance)
│   ├── services/
│   │   ├── keitaro_client.py        # Generic multi-instance Keitaro API client
│   │   ├── cache.py                 # Redis cache service (entity list caching)
│   │   └── crypto.py                # Fernet encrypt/decrypt/mask
│   └── modules/
│       ├── instances/               # Instance CRUD module
│       │   ├── router.py
│       │   ├── repository.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── keys/                    # API key management module
│       │   ├── router.py
│       │   ├── repository.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── campaigns/               # Campaign operations module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── offers/                  # Offer retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── landings/                # Landing page retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── domains/                 # Domain retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── conversions/             # Conversion log retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── reports/                 # Remote report building module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── traffic_sources/         # Traffic source retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── affiliate_networks/     # Affiliate network retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── users/                  # Keitaro tracker user retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── groups/                  # Group retrieval module
│       │   ├── router.py
│       │   ├── schemas.py
│       │   └── service.py
│       ├── health/                  # Health check module
│       │   ├── router.py
│       │   └── service.py
│       └── tasks/                   # Task status polling module
│           ├── router.py
│           └── schemas.py
├── migrations/                      # Alembic migrations
└── tests/                           # pytest tests