opengater-globalauth-client

Python client library for GlobalAuth authentication service.

Source release candidate: 0.7.1 (not published by this change).

Features

• 🔐 HTTP client for token validation and referral operations
• 🐰 RabbitMQ integration for events and commands
• ⚡ FastAPI middleware and dependencies
• 💾 Built-in caching for introspect results
• 🔄 Async-first design

Installation

pip install opengater-globalauth-client

Or with uv:

uv add opengater-globalauth-client

Quick Start

HTTP Client

import os
from opengater_globalauth_client import GlobalAuthClient

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="opengater",
    internal_api_key=os.environ.get("GLOBALAUTH_INTERNAL_API_KEY"),  # since 0.4.0
    cache_ttl=300  # Cache introspect results for 5 minutes
)

# Validate token
user = await client.introspect(token)
print(f"User: {user.id}, verified: {user.verified}")

# Decode referral code
ref_info = await client.decode_referral("ref_xxx")
if ref_info.type == "referral":
    print(f"Invited by: {ref_info.inviter_id}")

# Get referral link for user
link = await client.get_referral_link(user_token)
print(f"Share this link: {link.link}")

# Get referral stats
stats = await client.get_referral_stats(user_token)
print(f"Total invited: {stats.total_invited}")

# Auth method bonuses (inter-service, needs internal_api_key; since 0.5.1)
weights = await client.get_auth_method_weights()   # list[AuthMethodWeight]
profile = await client.get_auth_profile(user_id)   # AuthProfile
print(profile.registration_auth_type, profile.linked_auth_types)

RabbitMQ Integration

from faststream.rabbit import RabbitBroker
from opengater_globalauth_client.broker import (
    GlobalAuthConsumer,
    GlobalAuthPublisher,
    UserCreatedEvent,
    CreateUserCommand,
)

# Use your existing broker
broker = RabbitBroker(settings.rabbitmq_url)

# Consumer for events
consumer = GlobalAuthConsumer(broker, service_slug="opengater")

@consumer.on_user_created
async def handle_user_created(event: UserCreatedEvent):
    print(f"New user: {event.user_id}")
    
    if event.invited_by_id:
        # Award referral bonus
        await award_bonus(event.invited_by_id)

@consumer.on_auth_method_linked
async def handle_linked(event):
    print(f"User {event.user_id} linked {event.auth_type}")

service_slug is keyword-only and must be a plain slug (e.g. "opengater"), not a full queue name — passing something like "globalauth.opengater" (the pre-0.6.0 queue_name argument) raises ValueError.

At-most-once on handler errors. GlobalAuth's queues have no dead-letter exchange. With the default ack_policy=AckPolicy.REJECT_ON_ERROR, a handler that raises rejects the message and it is lost for good — not retried. Pass ack_policy=AckPolicy.NACK_ON_ERROR to requeue instead, but note the queues are classic with no delivery-count limit, so a handler that keeps failing on the same message will be retried forever (a poison-message loop).

# Publisher for commands (service_slug is required)
publisher = GlobalAuthPublisher(broker=broker, service_slug="opengater")

# Create user from Telegram bot
cmd = CreateUserCommand(
    auth_type="telegram",
    identifier="123456789",
    language="ru",
    extra_data={"username": "john", "first_name": "John"},
    invited_by_id="uuid-of-inviter",
)
await publisher.create_user(cmd)

Coordinated user deletion (0.7.0 candidate)

Deletion has separate typed consumer/publisher classes and does not change GlobalAuthConsumer or its REJECT_ON_ERROR default. The canonical v1 contract belongs to GlobalAuth. This source candidate requires the new GlobalAuth deletion coordinator; adding these classes does not make production deletion ready. Publish the versioned package only after separate maintainer approval, then update consumer locks to that actually published release.

from opengater_globalauth_client.broker import (
    GlobalAuthDeletionConsumer,
    GlobalAuthDeletionPublisher,
    ProjectCleanupRequested,
)

# Existing broker, connection credentials and RabbitMQ rights are reused.
consumer = GlobalAuthDeletionConsumer(
    broker,
    service_slug=settings.globalauth_service_slug,
    coordinator_dlx="users.dlx",  # Existing receiver-owned durable DIRECT exchange.
    delivery_limit=5,
)
publisher = GlobalAuthDeletionPublisher(
    broker, service_slug=settings.globalauth_service_slug,
)

async def accept_cleanup(command: ProjectCleanupRequested) -> None:
    # Must commit durable command/receipt before returning; verify the full
    # operation_id/step_id/target_user_id correlation in the coordinator.
    await coordinator.accept_cleanup(command)

consumer.on_cleanup_requested(accept_cleanup)
consumer.on_decision(coordinator.accept_decision)
consumer.on_finished(coordinator.accept_finished)

# Mandatory startup order, including every broker restart:
await consumer.prepare()  # Declare/bind every registered dead queue; errors abort startup.
await broker.start()

# The application saves outgoing DTOs before attempting publication.
# await publisher.request_deletion(saved_request)
# await publisher.publish_cleanup_result(saved_result)

service_slug is keyword-only and comes from trusted registered configuration. The exact registered string is preserved (including dots and uppercase). The constructor validates its local singleton; GlobalAuth registry/readiness must validate the full coordinator namespace before enabling it. Neither the queue name nor participant authenticates the publisher. Existing broker access is trusted in this version; AMQP user_id, new credentials/ACL and signatures are separate backlog work.

on_cleanup_requested(handler), on_decision(handler) and on_finished(handler) register asynchronous durable acceptance hooks and return None. Re-registering a phase replaces its hook while keeping one subscription. Register all hooks before preparation/startup. Adding a new phase invalidates preparation and requires another await consumer.prepare() before consuming. A hook's return never creates a CleanupResult or proves cleanup succeeded. Save the result and publish it explicitly after the application has finished its durable work.

prepare() declares durable dead queues without attaching consumers. Named coordinator_dlx exchanges are declared as durable DIRECT and explicitly bound; the default coordinator_dlx="" uses RabbitMQ's existing default exchange and its automatic binding, adding no named exchange. Any declaration/binding failure must abort startup/readiness. Work queues are durable quorum queues with x-delivery-limit (delivery_limit is a strict positive integer, default 5), x-dead-letter-exchange and an explicit dead routing key. Hook errors NACK/requeue within that finite transport limit; invalid/unsupported JSON and a mismatched cleanup participant are rejected to their .dead queue. An unprepared consumer fails without calling the hook or ACKing success. Business retries use saved failed/retryable results and coordinator delays rather than transport requeue. The receiver must verify quorum/delivery-limit/dead routing support in its live RabbitMQ before readiness; mock tests do not establish these broker guarantees.

Publication uses fixed default-exchange routing keys: globalauth.deletion.request.{service_slug} and globalauth.deletion.result.{service_slug}. Incoming subscriptions are exactly globalauth.deletion.{cleanup,decision,finished}.{service_slug}. JSON reply_to and source are forbidden by the DTOs; AMQP reply headers never trigger replies. The result participant must equal the publisher's configured slug. All four UUID identifiers and the UTC timestamp are preserved across wire round-trips/retries.

The publisher uses a separate FastStream-managed channel with confirms and unroutable returns enabled, preserving existing broker channel settings. Messages are persistent and mandatory, with a 10-second confirmation timeout. Only an actual Basic.Ack is accepted; return, NACK, timeout, no confirmation and network errors propagate. The application retains its outgoing as pending and retries the same DTO/IDs on failure. This transport confirmation does not prove the receiver's durable acceptance or business completion. The library provides no outbox, receipt storage or automatic cleanup logic.

DTOs (DeletionRequest, DeletionDecision, ProjectCleanupRequested, CleanupResult, DeletionFinished, DeletionMessage) and pure topology helpers are exported from the package root, .broker, and .broker.deletion. No new dependencies are introduced. Runtime tests of this candidate use Python 3.13; Python 3.11 has compile-only evidence until a supported dependency environment is available. Minimum FastStream 0.6.0's channel API was checked from its official wheel source; its runtime was not tested here.

Отчёты о проблемных шагах удаления (с 0.9.0)

Участник удаления не архивирует проблемные шаги у себя, а сообщает о них координатору проекта отчётом DeletionStepFailure. Координатор (в OpenGater — Users) объявляет и потребляет очередь отчётов, хранит их и реагирует; библиотека только формирует и доставляет отчёт.

• retrying — шаг передан в часовое ожидание <очередь>.retry.hour; код deletion_temporarily_unavailable; error — только имя класса исключения; тело сообщения не передаётся.
• dead — сообщение попало в .dead, корреляция известна; код deletion_attempts_exhausted (x-death.reason = delivery_limit) или deletion_command_rejected (иные причины); error = death_reason=<reason>.
• uncorrelated — тело из .dead разобрать нельзя; код deletion_message_uncorrelated, operation_id пустой, сырое тело сохраняется.

raw_body декодируется как UTF-8 с заменой некорректных байтов, символ NUL заменяется на U+FFFD (так же в error, source_queue, message_id), затем тело обрезается до 16384 символов. report_id детерминирован: повторная доставка того же отчёта в том же цикле даёт тот же идентификатор, и координатор обрабатывает её идемпотентно.

from opengater_globalauth_client.broker import (
    DeletionDeadLetterHandler,
    DeletionFailureReporter,
    ProjectCleanupRequested,
    build_retrying_failure,
    correlation_parser,
)
from faststream.rabbit import RabbitQueue

reporter = DeletionFailureReporter(broker)  # очередь users.deletion.failure
parse = correlation_parser(ProjectCleanupRequested)  # модель команды шага

# Потребитель .dead: отчёт dead/uncorrelated, затем ACK. Очередь .dead объявляет
# участник, как и раньше (аргументы не меняются).
DeletionDeadLetterHandler(
    reporter, participant="payments", source_queue="payments.deletion.request", parse=parse,
).subscribe(broker, RabbitQueue("payments.deletion.request.dead", durable=True))

# Временная ошибка шага: сначала подтверждённая передача в .retry.hour, потом отчёт.
await publish_to_retry_hour(message)  # код сервиса, с publisher confirm
await reporter.report(build_retrying_failure(
    participant="payments",
    source_queue="payments.deletion.request",
    correlation=parse(message.body),
    body=message.body,
    message_id=message.raw_message.message_id,
    headers=message.headers,
    error=exc,
))

Отчёт публикуется через default exchange: persistent, mandatory, publisher confirm (10 секунд) на отдельном канале. Без настоящего Basic.Ack — DeletionFailureReportUnconfirmed; возврат mandatory (очереди ещё нет) и сетевые ошибки пробрасываются. DeletionDeadLetterHandler подписывается с AckPolicy.MANUAL и prefetch 1 и подтверждает сообщение .dead только после confirm отчёта. Если отчёт отправить не удалось, сообщение остаётся неподтверждённым (без ack/nack/reject), в лог пишется предупреждение без тела и текста исключения; потребление этой очереди приостанавливается до восстановления канала или перезапуска сервиса, после чего RabbitMQ доставит сообщение снова. Поэтому координатор должен объявить очередь отчётов раньше, чем участники начнут их отправлять.

Выкатка: участник, запущенный раньше, чем координатор объявил очередь отчётов, держит сообщение .dead неподтверждённым до перезапуска или восстановления канала — перезапустите участников после выкатки Users.

Любая ошибка parse (не только ValidationError/ValueError) превращает отчёт в uncorrelated; неподтверждённым сообщение остаётся только при сбое публикации отчёта. participant и source_queue проверяются в конструкторе обработчика.

message_id берите из AMQP-свойства (message.raw_message.message_id), а не из message.message_id FastStream: при отсутствии свойства FastStream подставляет случайный идентификатор, и report_id перестал бы быть детерминированным.

GlobalAuth не зависит от этой библиотеки и держит зеркальную модель; сверка — по tests/fixtures/deletion_failure_v1.json — набор векторов (vectors: dead с message_id и без него, uncorrelated, retrying), у каждого входы построителя, входы и значение report_id и точное тело сообщения. Свои отчёты GlobalAuth публикует в брокер проекта в очередь globalauth_failure_queue_name(coordinator_slug) = globalauth.deletion.failure.<slug>: DeletionFailureReporter(broker, queue_name=globalauth_failure_queue_name("opengater-users")).

FastAPI Integration

With Middleware (Recommended)

from fastapi import FastAPI, Depends, Request
from opengater_globalauth_client import GlobalAuthClient
from opengater_globalauth_client.fastapi import AuthMiddleware, get_current_user

app = FastAPI()

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="opengater"
)

# Add middleware for automatic token validation
app.add_middleware(
    AuthMiddleware,
    client=client,
    exclude_paths=["/health", "/public/*"]  # Optional
)

@app.get("/me")
async def me(user = Depends(get_current_user())):
    return {
        "id": user.id,
        "verified": user.verified,
        "trust_level": user.trust.level
    }

@app.get("/admin-only")
async def admin_only(user = Depends(require_admin())):
    return {"message": "Welcome, admin!"}

Without Middleware

from fastapi import FastAPI, Depends
from opengater_globalauth_client import GlobalAuthClient
from opengater_globalauth_client.fastapi import get_current_user

app = FastAPI()

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="opengater"
)

# Pass client to dependency
@app.get("/me")
async def me(user = Depends(get_current_user(client))):
    return {"id": user.id}

Role checks (since 0.8.0)

from opengater_globalauth_client import Role
from opengater_globalauth_client.fastapi import (
    require_ad_manager, require_editor, require_roles,
)

@app.post("/articles")
async def create_article(user = Depends(require_editor())):  # admin, editor
    ...

@app.post("/campaigns")
async def create_campaign(user = Depends(require_ad_manager())):  # admin, ad_manager
    ...

@app.post("/custom")
async def custom(user = Depends(require_roles(Role.ADMIN, "moderator"))):
    ...  # допускаются только перечисленные роли; admin неявно не допускается

Role — перечень известных ролей, но User.role остаётся строкой: роль, добавленная в GlobalAuth позже, не ломает разбор. Остальным ролям — 403.

Available Events

User Model

class User:
    id: str
    verified: bool
    is_banned: bool
    role: str  # см. Role: user, support, admin, editor, ad_manager; неизвестные значения допустимы
    trust: TrustInfo  # score, level
    auth_methods: list[AuthMethod]
    invited_by_id: str | None
    registered_via_event: str | None
    registered_via_service: str | None

Configuration

Inter-service API key (since 0.4.0)

client = GlobalAuthClient(
    base_url="https://auth.example.com",
    service_slug="my-service",
    internal_api_key=os.environ.get("GLOBALAUTH_INTERNAL_API_KEY"),
)

internal_api_key (optional) — inter-service API ключ. Когда задан, отправляется как X-API-Key на /auth/jwt/introspect. Нужен начиная с GlobalAuth E5 (hard requirement). До E5 — игнорируется сервером.

Error Handling

from opengater_globalauth_client import (
    GlobalAuthClient,
    TokenExpiredError,
    TokenInvalidError,
    UserBannedError,
    GlobalAuthConnectionError,
)

try:
    user = await client.introspect(token)
except TokenExpiredError:
    # Token expired, need to refresh
    pass
except UserBannedError:
    # User is banned
    pass
except TokenInvalidError:
    # Invalid token
    pass
except GlobalAuthConnectionError:
    # GlobalAuth service unavailable
    pass

Development

This package uses uv for dependency management.

# Install dev dependencies
uv sync --group dev

# Run tests
uv run pytest

# Lint & type-check
uv run ruff check src tests
uv run mypy src

# Build a release
uv build  # → dist/opengater_globalauth_client-X.Y.Z-*.whl

Publishing to PyPI (maintainer-only):

uv build
twine upload dist/opengater_globalauth_client-X.Y.Z*

License

MIT