aiogram-keycloak-sso

SSO (OpenID Connect / Keycloak) authentication middleware for aiogram 3: outer/inner middlewares, handler flags, DI through the data dict, and a swappable storage protocol.

This library is the bot-side adapter (login URL, session middleware, consume_login_events). Completing the OIDC browser callback is pluggable: any backend that writes the session and publishes a LoginEvent on the shared event bus works. Wire DTOs (SSOSession, login events, Redis prefixes) must stay in sync with that backend (schema_version).

Callback backends (pick one)

Recommended for fleets: keep the callback in a separate process (gateway or your own). For a single bot / lab setup, attach an inline listener:

from aiogram_keycloak_sso import CallbackHttpListener, FleetBotManager, MemoryEventBus

bus = MemoryEventBus()  # or RedisEventBus
listener = CallbackHttpListener(event_bus=bus)
manager.add_listener(listener)          # bind only — no server yet
listener.mount(existing_aiohttp_app)    # or: await listener.start(port=8080)

How it maps onto aiogram's design

Install

pip install aiogram-keycloak-sso
pip install "aiogram-keycloak-sso[redis]"   # Redis sessions + event bus

Import package name (underscore): aiogram_keycloak_sso.

Quick start (bot process + external callback)

from aiogram import Bot, Dispatcher, Router
from aiogram.filters import Command
from redis.asyncio import Redis

from aiogram_keycloak_sso import (
    AccessPolicy,
    FleetBotManager,
    SSOConfig,
    SSOSession,
    SSOSessionMiddleware,
    SSOAccessMiddleware,
    sso_required,
    login_notifier,
    login_failed_notifier,
    login_denied_notifier,
    consume_login_events,
)
from aiogram_keycloak_sso.events_redis import RedisEventBus
from aiogram_keycloak_sso.storage.redis import RedisSessionStorage

POLICY = AccessPolicy(groups=["/bots/alpha"])

redis = Redis.from_url("redis://localhost:6379/0")
config = SSOConfig(
    issuer="https://keycloak.example.com/realms/main",
    client_id="my-bot",
    client_secret="...",
    redirect_uri="https://sso.example.com/sso/callback",  # callback public URL
    state_secret="long-random-string-keep-it-secret",     # same as on the callback backend
    refresh_leeway=30,
)
manager = FleetBotManager("alpha", config, RedisSessionStorage(redis))

bot = Bot(token="...")
dp = Dispatcher()
router = Router()
dp.update.outer_middleware(SSOSessionMiddleware(manager, policy=POLICY))
router.message.middleware(SSOAccessMiddleware())
dp.include_router(router)

@router.message(Command("me"))
@sso_required
async def me(message, sso_session: SSOSession):
    await message.answer(f"{sso_session.username} / roles: {sso_session.realm_roles}")

@router.message(Command("admin"))
@sso_required(realm_roles=["admin"])
async def admin(message, sso_session: SSOSession): ...

await consume_login_events(
    RedisEventBus(redis),
    "alpha",
    on_login=login_notifier(bot),
    on_login_failed=login_failed_notifier(bot),
    policy=POLICY,
    on_denied=login_denied_notifier(bot),
)
# then: await dp.start_polling(bot)

Runnable example: examples/bot_alpha.py. Required process env vars are listed in .env.example (example only — the package does not load .env).

After a successful login

from aiogram_keycloak_sso import login_notifier, login_failed_notifier, login_denied_notifier

await consume_login_events(
    bus, "alpha",
    on_login=login_notifier(bot),                # "✅ Signed in as {display_name}"
    on_login_failed=login_failed_notifier(bot),  # "❌ Sign-in was not completed..."
    policy=POLICY,
    on_denied=login_denied_notifier(bot),        # "❌ ... no access to this bot"
)

Chat hooks live on consume_login_events (event bus). The callback backend — gateway, custom service, or CallbackHttpListener — writes the session and publishes the event. Same-process setups can also pass on_login / on_login_failed into SSOManager / FleetBotManager directly.

Localization

Built-in strings (en / ru) are localizable; every component accepts locale:

SSOAccessMiddleware()                    # locale="auto" → Telegram language_code
SSOAccessMiddleware(locale="ru-RU")      # pin one language
login_notifier(bot)                      # auto: IdP locale claim, then Telegram

Add a language with register_texts("de", SSOTexts(...)).

Continue where the user left off (next)

SSOAccessMiddleware captures the interrupted action into signed state. After login it returns as result.state.next for a "continue" button.

Session lifecycle & TTL

• TTL defaults to the access token's expires_in; session_ttl caps it.
• Outer middleware refreshes via the refresh grant when TTL elapses (minus refresh_leeway). Concurrent updates for one user refresh once.
• Rejected refresh → session dropped → protected handlers show a login button.
• manager.logout(user_id) revokes the refresh token (RFC 7009) and deletes the session.

Optional claims enricher

Prefer putting org attributes in Keycloak / FreeIPA claims. If you need an extra directory lookup, pass a SessionEnricher into FleetBotManager / SSOManager — it runs on login (handle_callback) and again on each OIDC token refresh.

Security notes

• state is HMAC-SHA256-signed (telegram user id, timestamp, nonce).
• OIDC ID-token nonce is optional (require_id_token_nonce=False by default); enable it only if your IdP echoes nonce into the ID token.
• ID tokens validated against realm JWKS (iss, aud, exp).

Keycloak setup

1. Confidential client, Standard flow enabled.
2. Add the callback URL to Valid redirect URIs (one URI for the fleet).
3. Role mappers with Add to ID token enabled (realm_access.roles / resource_access.*.roles). Verify with Client scopes → Evaluate.

Multi-bot fleets

Wire contracts (duplicated across bot + callback packages; keep in sync via schema_version):

Isolation is derived from bot_id on both sides (storage namespace, event channel, derived state secret). If the callback backend is down, new logins pause; bots keep serving live sessions from storage.

Access control (AccessPolicy)

When per-bot Keycloak clients are not an option, authorize on the consumer:

from aiogram_keycloak_sso import AccessPolicy

POLICY = AccessPolicy(
    groups=["/bots/alpha"],
    claims={"department": ["R&D", "IT"]},
)

dp.update.outer_middleware(SSOSessionMiddleware(manager, policy=POLICY))
await consume_login_events(
    bus, "alpha",
    on_login=login_notifier(bot),
    policy=POLICY,
    on_denied=login_denied_notifier(bot),
)

Compose with | / & (RND | INVITED). Same object works with @sso_required(POLICY).

Custom storage

Implement BaseSessionStorage (get / set / delete). Redis backend: aiogram_keycloak_sso.storage.redis (requires [redis] extra).

Development

git clone https://github.com/ggb0yyyyyyy/sso-aiogram-midleware.git
cd sso-aiogram-midleware
pip install -e ".[dev,redis]"
pytest

License

MIT — see LICENSE.