better time usings
This commit is contained in:
@@ -12,7 +12,7 @@ can call it from async routes via threadpool if needed.
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from datetime import datetime, timedelta
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Dict, Optional
|
||||
|
||||
from jose import JWTError, jwt # type: ignore
|
||||
@@ -103,10 +103,10 @@ class AuthService:
|
||||
def _record_failure(self, identifier: str) -> None:
|
||||
rec = self._get_fail_record(identifier)
|
||||
rec["count"] += 1
|
||||
rec["last"] = datetime.utcnow()
|
||||
rec["last"] = datetime.now(timezone.utc)
|
||||
if rec["count"] >= self.max_attempts:
|
||||
rec["locked_until"] = (
|
||||
datetime.utcnow() + timedelta(seconds=self.lockout_seconds)
|
||||
datetime.now(timezone.utc) + timedelta(seconds=self.lockout_seconds)
|
||||
)
|
||||
|
||||
def _clear_failures(self, identifier: str) -> None:
|
||||
@@ -116,11 +116,11 @@ class AuthService:
|
||||
def _check_locked(self, identifier: str) -> None:
|
||||
rec = self._get_fail_record(identifier)
|
||||
lu = rec.get("locked_until")
|
||||
if lu and datetime.utcnow() < lu:
|
||||
if lu and datetime.now(timezone.utc) < lu:
|
||||
raise LockedOutError(
|
||||
"Too many failed attempts - temporarily locked out"
|
||||
)
|
||||
if lu and datetime.utcnow() >= lu:
|
||||
if lu and datetime.now(timezone.utc) >= lu:
|
||||
# lock expired, reset
|
||||
self._failed[identifier] = {
|
||||
"count": 0,
|
||||
@@ -155,13 +155,13 @@ class AuthService:
|
||||
def create_access_token(
|
||||
self, subject: str = "master", remember: bool = False
|
||||
) -> LoginResponse:
|
||||
expiry = datetime.utcnow() + timedelta(
|
||||
expiry = datetime.now(timezone.utc) + timedelta(
|
||||
hours=(168 if remember else self.token_expiry_hours)
|
||||
)
|
||||
payload = {
|
||||
"sub": subject,
|
||||
"exp": int(expiry.timestamp()),
|
||||
"iat": int(datetime.utcnow().timestamp()),
|
||||
"iat": int(datetime.now(timezone.utc).timestamp()),
|
||||
}
|
||||
token = jwt.encode(payload, self.secret, algorithm="HS256")
|
||||
|
||||
@@ -180,7 +180,7 @@ class AuthService:
|
||||
data = self.decode_token(token)
|
||||
exp_val = data.get("exp")
|
||||
expires_at = (
|
||||
datetime.utcfromtimestamp(exp_val) if exp_val is not None else None
|
||||
datetime.fromtimestamp(exp_val, timezone.utc) if exp_val is not None else None
|
||||
)
|
||||
return SessionModel(
|
||||
session_id=hashlib.sha256(token.encode()).hexdigest(),
|
||||
|
||||
Reference in New Issue
Block a user