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.

Bots never expose HTTP. The OIDC callback runs in a companion process — keycloak-sso-gateway (separate package / repo). This library is the bot-side adapter only (polling / outbound IdP access). Wire DTOs (SSOSession, login events, Redis prefixes) are duplicated in both packages and must stay in sync (schema_version).

 Telegram user           bot (aiogram-keycloak-sso)         callback service         IdP
──────────────          ─────────────────         ────────────────         ──────
 /login  ──────────────▶ FleetBotManager
                         state = bot:<bot_id>:signed
 opens URL ──────────────────────────────────────────────────────────────────▶ login
                                                   ◀── redirect ── code+state
                                                   code → tokens, store Redis
                                                   publish login event
                         consume_login_events ◀─── Redis pub/sub
 any update ───────────▶ SSOSessionMiddleware
                         (session from Redis; refresh ───────────────────────▶)
 flagged handler ──────▶ SSOAccessMiddleware + @sso_required

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.

The HTTP callback service is not included — run keycloak-sso-gateway as a separate process (python -m keycloak_sso_gateway / Docker Compose).

Quick start (bot process)

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",  # gateway public URL
    state_secret="long-random-string-keep-it-secret",     # same as on the gateway
    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)  — no inbound HTTP

Runnable example: examples/bot_alpha.py (env template: .env.example).

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"
)

Login chat hooks live on consume_login_events (Redis pub/sub from the gateway). The bot does not handle the OIDC HTTP callback.

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 on the bot, pass a SessionEnricher into FleetBotManager / SSOManager — it runs again on each OIDC token refresh. Initial enrichment at login belongs in the gateway (it writes the session).

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 gateway 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 in both 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 gateway is down, new logins pause; bots keep serving live Redis sessions.

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.