damanukyan/adminka
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add import

Browse Source
This commit is contained in:
Daniil Manukian
2026-06-04 18:07:08 +03:00
parent 4eb2c78c03
commit 62a48c3f70
13 changed files with 104 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/infrastructure/security/__init__.py
View File

@@ -1,3 +1,2 @@
from src.infrastructure.security.jwt import JwtService
from src.infrastructure.security.csrf import CsrfService
from src.infrastructure.security.hash import HashService

81
src/infrastructure/security/csrf.py
View File

@@ -1,81 +0,0 @@
from __future__ import annotations
import secrets
from typing import Any, Optional, Mapping
from itsdangerous import URLSafeTimedSerializer, SignatureExpired, BadSignature
from src.application.contracts import ICsrfService
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config.settings import settings
class CsrfService(ICsrfService):
COOKIE_NAME = "csrf_token"
HEADER_NAME = "X-CSRF-Token"
SALT = "csrf"
TTL_SECONDS = 3600
def __init__(self) -> None:
self._serializer = URLSafeTimedSerializer(
secret_key=settings.CSRF_SECRET_KEY,
salt=self.SALT,
)
@property
def cookie_name(self) -> str:
return self.COOKIE_NAME
@property
def header_name(self) -> str:
return self.HEADER_NAME
@property
def ttl_seconds(self) -> int:
return self.TTL_SECONDS
def issue(self, subject: Optional[str] = None) -> str:
payload = {
"sub": subject,
"nonce": secrets.token_urlsafe(32),
}
return self._serializer.dumps(payload)
def verify(self, token: str, expected_subject: Optional[str] = None) -> dict[str, Any]:
try:
data = self._serializer.loads(token, max_age=self.TTL_SECONDS)
except SignatureExpired:
raise ApplicationException(
status_code=403,
message="CSRF token expired",
)
except BadSignature:
raise ApplicationException(
status_code=403,
message="CSRF token invalid",
)
if expected_subject is not None and data.get("sub") != expected_subject:
raise ApplicationException(
status_code=403,
message="CSRF token subject mismatch",
)
return data
def extract(self, cookies: Mapping[str, str], headers: Mapping[str, str]) -> tuple[Optional[str], Optional[str]]:
cookie_token = cookies.get(self.COOKIE_NAME)
header_token = headers.get(self.HEADER_NAME)
return cookie_token, header_token
def verify_pair(self, cookie_token: Optional[str], header_token: Optional[str], expected_subject: Optional[str] = None) -> None:
if not cookie_token or not header_token:
raise ApplicationException(
status_code=403,
message="CSRF token missing",
)
if not secrets.compare_digest(cookie_token, header_token):
raise ApplicationException(
status_code=403,
message="CSRF token mismatch",
)
self.verify(cookie_token, expected_subject=expected_subject)

58
src/infrastructure/security/jwt.py
View File

@@ -5,7 +5,7 @@ from datetime import datetime, timezone, timedelta
from jose import jwt, ExpiredSignatureError, JWTError
from src.application.contracts import ILogger, IJwtService
from src.application.domain.dto import AccessTokenPayload
from src.application.domain.dto import AccessTokenPayload, RefreshTokenPayload
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config.settings import settings
from src.infrastructure.vault import JwtKeyStore
@@ -20,7 +20,7 @@ class JwtService(IJwtService):
def _issuer(self) -> str | None:
return settings.ADMIN_JWT_ISSUER
async def create_access_token(self, user_id: str, *, role: str) -> str:
async def create_access_token(self, user_id: str, *, role: str, sid: str | None = None) -> str:
now = datetime.now(timezone.utc)
exp = now + timedelta(seconds=int(settings.JWT_ACCESS_TTL_SECONDS))
@@ -32,15 +32,40 @@ class JwtService(IJwtService):
'nbf': int(now.timestamp()),
'exp': int(exp.timestamp()),
}
if sid:
payload['sid'] = sid
if self._issuer:
payload['iss'] = self._issuer
if settings.JWT_AUDIENCE:
payload['aud'] = settings.JWT_AUDIENCE
return await self._encode(payload, user_id=user_id, token_kind='access')
async def create_refresh_token(self, user_id: str, *, sid: str, refresh_jti: str) -> str:
now = datetime.now(timezone.utc)
exp = now + timedelta(seconds=int(settings.JWT_REFRESH_TTL_SECONDS))
payload: dict[str, object] = {
'sub': user_id,
'type': 'refresh',
'sid': sid,
'jti': refresh_jti,
'iat': int(now.timestamp()),
'nbf': int(now.timestamp()),
'exp': int(exp.timestamp()),
}
if self._issuer:
payload['iss'] = self._issuer
if settings.JWT_AUDIENCE:
payload['aud'] = settings.JWT_AUDIENCE
return await self._encode(payload, user_id=user_id, token_kind='refresh')
async def _encode(self, payload: dict[str, object], *, user_id: str, token_kind: str) -> str:
try:
kid, private_pem = await self._key_store.get_signing_key()
token = jwt.encode(payload, private_pem, algorithm=settings.JWT_ALGORITHM, headers={'kid': kid})
self._logger.info(f'Admin access token created admin_user_id={user_id} kid={kid}')
self._logger.info(f'Admin {token_kind} token created admin_user_id={user_id} kid={kid}')
return token
except ApplicationException:
raise
@@ -57,6 +82,26 @@ class JwtService(IJwtService):
sub=str(payload['sub']),
type='access',
role=str(payload['role']) if payload.get('role') else None,
sid=str(payload['sid']) if payload.get('sid') else None,
iat=int(payload['iat']),
nbf=int(payload['nbf']),
exp=int(payload['exp']),
iss=payload.get('iss'),
aud=payload.get('aud'),
)
except KeyError as exception:
raise ApplicationException(status_code=401, message=f'Missing token claim: {exception}')
async def decode_refresh_token(self, token: str) -> RefreshTokenPayload:
payload = await self._decode_and_verify(token)
if payload.get('type') != 'refresh':
raise ApplicationException(status_code=401, message='Invalid token type')
try:
return RefreshTokenPayload(
sub=str(payload['sub']),
type='refresh',
sid=str(payload['sid']),
jti=str(payload['jti']),
iat=int(payload['iat']),
nbf=int(payload['nbf']),
exp=int(payload['exp']),
@@ -104,8 +149,13 @@ class JwtService(IJwtService):
)
if 'type' not in payload:
raise ApplicationException(status_code=401, message='Missing token claim: type')
if payload.get('type') == 'access' and 'role' not in payload:
token_type = payload.get('type')
if token_type == 'access' and 'role' not in payload:
raise ApplicationException(status_code=401, message='Missing token claim: role')
if token_type == 'refresh':
for claim in ('sid', 'jti'):
if claim not in payload:
raise ApplicationException(status_code=401, message=f'Missing token claim: {claim}')
return payload
except ExpiredSignatureError:
raise ApplicationException(status_code=401, message='Token expired')