84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
import hashlib
|
|
import hmac
|
|
import secrets
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
import jwt
|
|
|
|
from gapido_auth.domain.entities import AccessClaims, Role
|
|
from gapido_auth.domain.errors import AuthenticationError
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def generate_otp_code() -> str:
|
|
return f"{secrets.randbelow(1_000_000):06d}"
|
|
|
|
|
|
def hash_otp(secret: str, mobile: str, purpose: str, code: str) -> str:
|
|
message = f"{mobile}:{purpose}:{code}".encode()
|
|
return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
|
|
|
|
|
|
def generate_refresh_token() -> str:
|
|
return secrets.token_urlsafe(48)
|
|
|
|
|
|
def hash_refresh_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
|
|
class JwtTokenCodec:
|
|
def __init__(self, secret_key: str, issuer: str, access_ttl_seconds: int) -> None:
|
|
self._secret_key = secret_key
|
|
self._issuer = issuer
|
|
self._access_ttl_seconds = access_ttl_seconds
|
|
|
|
@property
|
|
def access_ttl_seconds(self) -> int:
|
|
return self._access_ttl_seconds
|
|
|
|
def encode_access_token(self, user_id: str, role: Role) -> str:
|
|
now = utc_now()
|
|
expires_at = now + timedelta(seconds=self._access_ttl_seconds)
|
|
payload = {
|
|
"iss": self._issuer,
|
|
"sub": user_id,
|
|
"role": role.value,
|
|
"typ": "access",
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expires_at.timestamp()),
|
|
"jti": str(uuid4()),
|
|
}
|
|
return jwt.encode(payload, self._secret_key, algorithm="HS256")
|
|
|
|
def decode_access_token(self, token: str) -> AccessClaims:
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
self._secret_key,
|
|
algorithms=["HS256"],
|
|
issuer=self._issuer,
|
|
options={"require": ["exp", "iat", "iss", "sub", "role", "typ"]},
|
|
)
|
|
except jwt.PyJWTError as exc:
|
|
raise AuthenticationError("invalid access token") from exc
|
|
|
|
if payload.get("typ") != "access":
|
|
raise AuthenticationError("invalid token type")
|
|
|
|
try:
|
|
role = Role(str(payload["role"]))
|
|
except ValueError as exc:
|
|
raise AuthenticationError("invalid role claim") from exc
|
|
|
|
return AccessClaims(
|
|
user_id=str(payload["sub"]),
|
|
role=role,
|
|
expires_at=datetime.fromtimestamp(int(payload["exp"]), UTC),
|
|
)
|
|
|