s-authkit-server

СЕРВЕРНОЕ ядро авторизации для Skillery проектов: JWT RS256, refresh-token management, Argon2 hashing. Импорт — authkit_server.

Версия: 0.1.1
Лицензия: MIT
Зависимости: python-jose[cryptography], passlib[argon2], cryptography

Какая это сторона

Кит	Импорт	Чью авторизацию решает
s-authkit-server (этот)	authkit_server	НАШ сервис авторизует СВОИХ пользователей: свои токены, свои пароли, свои refresh
s-authkit-client	authkit_client	НАШ код получает доступ к ЧУЖИМ сервисам: сессии, секреты, OAuth-обновление, живая проба

Суффикс имени и есть указатель стороны. До 0.1.1 этот кит назывался s-authkit (импорт authkit) — из-за пары «authkit / authkit-client» серверный кит читался как SDK к клиентскому. См. раздел «Переименование» ниже.

Что входит

JWT access-токены (RS256)

• JoseTokenIssuer — выпуск и верификация JWT
• TokenPair / TokenClaims — value objects для типизации
• ensure_jwt_keypair — генерация RSA 2048 ключей

Persistent refresh-tokens

• IRefreshTokenStore — protocol для БД-реализации
• RefreshTokenRecord — агрегат с хешированием и валидацией
• Rotation с audit-trail (rotated_from)

Password hashing

• Argon2PasswordHasher — Argon2id через passlib

API-ключи (для машин)

Способ по умолчанию, когда к сервису ходит не человек, а устройство или другой сервис. Ключ живёт долго и отзывается поштучно: потеряли ноутбук — отозвали его ключ, остальные машины работают.

from authkit_server import ApiKeyService

ключи = ApiKeyService(store=МойСтор())
выпуск = ключи.issue(subject="pc-home", scopes=("totp:read",), label="ПК дома")
print(выпуск.plaintext)   # ak_9f2c…_KZ8… — показывается ОДИН раз

запись = ключи.verify(предъявленный)   # ApiKeyError, если не годится
запись.has_scope("totp:read")          # True; "totp:*" покрывает семейство
ключи.revoke(запись.key_id)

TOTP (второй фактор)

Серверная сторона: наш сервис заводит секрет своему пользователю и проверяет присланные коды. Обратная задача — хранить чужие секреты и предъявлять коды Битриксу или PyPI — решается китом totpkit.

from authkit_server import TotpVerifier, provision_totp

выпуск = provision_totp(account="dmitry", issuer="Cifro")
выпуск.uri        # otpauth://… — показать QR-кодом

проверка = TotpVerifier(guard=МойСторОкон())
проверка.verify(subject="dmitry", secret=выпуск.secret, code="123456")

ReplayGuard обязателен: без него подсмотренный код работает до конца тридцатисекундного окна.

Единый интерфейс способов входа

Сервис объявляет, что принимает, и получает Principal — кто пришёл и с какими правами. Чем именно представился клиент, сервису знать не нужно.

from authkit_server import ApiKeyScheme, Authenticator, JwtScheme

вход = Authenticator(schemes=[ApiKeyScheme(ключи), JwtScheme(издатель)])
кто = вход.authenticate(предъявленное, require="totp:read")
кто.subject, кто.method   # ("pc-home", "api_key")

Добавление нового способа — ещё один элемент списка; сервисы, живущие на ключах, не меняются.

Примеры

1. Инициализация

from pathlib import Path
from authkit_server import (
    JoseTokenIssuer,
    Argon2PasswordHasher,
    ensure_jwt_keypair,
)

# Создаём ключи (если нет)
keys_dir = Path.home() / ".myapp" / "keys"
ensure_jwt_keypair(
    private_path=keys_dir / "jwt_private.pem",
    public_path=keys_dir / "jwt_public.pem",
)

# Прочитаем ключи
private_key = (keys_dir / "jwt_private.pem").read_text()
public_key = (keys_dir / "jwt_public.pem").read_text()

# Создаём компоненты
hasher = Argon2PasswordHasher()
issuer = JoseTokenIssuer(
    private_key_pem=private_key,
    public_key_pem=public_key,
    issuer="myproject.com",
    access_ttl_min=15,
    refresh_ttl_days=30,
    refresh_store=your_store_impl,  # реализуете вы
)

2. Login (выпуск пары токенов)

async def login_with_password(username: str, password: str):
    # Получаем юзера из БД
    user = await db.get_user_by_username(username)
    if not user:
        raise ValueError("User not found")
    
    # Проверяем пароль
    if not hasher.verify(password, user.password_hash):
        raise ValueError("Invalid password")
    
    # Выпускаем пару
    pair = await issuer.issue_pair(
        user_id=str(user.id),
        company_id=str(user.active_company_id) if user.active_company_id else None,
        permissions={"skill.read", "skill.install"},
    )
    return pair

3. Middleware (верификация токена)

from authkit_server import TokenError

async def auth_middleware(request, call_next):
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        return Response("Unauthorized", status_code=401)
    
    token = auth_header[7:]
    try:
        claims = issuer.verify_access(token)
    except TokenError as e:
        return Response(f"Invalid token: {e}", status_code=401)
    
    request.state.claims = claims
    return await call_next(request)

4. Refresh (rotation)

async def refresh_session(refresh_token: str):
    try:
        new_pair = await issuer.rotate_refresh(refresh_token)
    except RefreshTokenError as e:
        raise Unauthorized(f"Refresh failed: {e}")
    return new_pair

5. Реализация IRefreshTokenStore

from authkit_server import RefreshTokenRecord, IRefreshTokenStore

class PostgresRefreshTokenStore:
    def __init__(self, db_engine):
        self.engine = db_engine
    
    async def save(self, record: RefreshTokenRecord) -> None:
        # INSERT/UPDATE в БД
        async with self.engine.begin() as conn:
            await conn.execute(
                "INSERT INTO refresh_tokens (user_id, token_hash, ...) VALUES (...)"
            )
    
    async def get_by_hash(self, token_hash: str) -> RefreshTokenRecord | None:
        # SELECT * FROM refresh_tokens WHERE token_hash = ?
        ...
    
    async def revoke_all_for_user(self, user_id: str) -> int:
        # UPDATE refresh_tokens SET revoked_at = NOW() WHERE user_id = ?
        ...
    
    # Реализуете остальные методы Protocol'а

Архитектура

authkit_server/
├── __init__.py              # Public API
├── exceptions.py            # AuthKitError, TokenError, ...
├── token/
│   ├── models.py            # TokenPair, TokenClaims
│   ├── jose_issuer.py       # JoseTokenIssuer
│   └── keys.py              # ensure_jwt_keypair
├── refresh/
│   └── models.py            # RefreshTokenRecord, IRefreshTokenStore
└── password/
    └── hasher.py            # Argon2PasswordHasher

Тестирование

pytest tests/                      # Все тесты
pytest tests/ -v --cov           # С coverage report
pytest tests/ -k "test_verify"   # Конкретный тест

Coverage gate: ≥ 80%.

Что НЕ входит (v0.1)

• OAuth (YandexOAuth, ExchangeCode) — в v0.2+
• PermissionResolver — остаются на стороне приложения
• Redis cache invalidation — в v0.2+ как optional
• Signing PK/SK rotation — будущая фича
• Sync обёртки — только async API в v0.1

Интеграция в существующее приложение

Типовой путь перевода существующего приложения на модуль:

1. Импортирует from authkit_server import ... вместо локального кода
2. Обёрнет RefreshTokenRecord в ORM-адаптер для своей БД
3. Оставит PermissionResolver/OAuth/ExchangeCode локально

Переименование (0.1.0 → 0.1.1)

Кит переименован: дистрибутив s-authkit → s-authkit-server, импорт authkit → authkit_server. Публичный API (имена классов и функций) не менялся — правится только строка импорта и строка зависимости.

Со старым дистрибутивом на PyPI:

• s-authkit==0.1.0 остаётся опубликованным как есть — его не отзываем (yank) и не удаляем: он рабочий, а отзыв сломал бы любую уже собранную сборку;
• новых выпусков под именем s-authkit больше не будет — имя заморожено на 0.1.0;
• переименование на PyPI «на месте» невозможно: s-authkit-server — это НОВЫЙ дистрибутив, первый его выпуск — 0.1.1 (нумерация продолжает историю кита, а не начинается заново, чтобы версия читалась как «то же ядро, новое имя»).

Миграция потребителя — две строки:

-"s-authkit>=0.1.0",
+"s-authkit-server>=0.1.1",

-from authkit import Argon2PasswordHasher, ensure_jwt_keypair
+from authkit_server import Argon2PasswordHasher, ensure_jwt_keypair

Лицензия

MIT