s-authkit

Ядро авторизации для Skillery проектов: JWT RS256, refresh-token management, Argon2 hashing.

Версия: 0.1 (первый релиз)
Лицензия: MIT
Зависимости: python-jose[cryptography], passlib[argon2], cryptography

Что входит

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

Примеры

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

from pathlib import Path
from authkit 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 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 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/
├── __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 import ... вместо локального кода
2. Обёрнет RefreshTokenRecord в ORM-адаптер для своей БД
3. Оставит PermissionResolver/OAuth/ExchangeCode локально

Лицензия

MIT