opengater-globalauth-client

Python client library for GlobalAuth authentication service.

Source release candidate: 0.7.0 (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.

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}

Available Events

User Model

class User:
    id: str
    verified: bool
    is_banned: bool
    role: Literal["user", "support", "admin"]
    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