Spaces:
Running
Running
Update app/auth.py
Browse files- app/auth.py +17 -9
app/auth.py
CHANGED
|
@@ -1,23 +1,23 @@
|
|
| 1 |
import os
|
|
|
|
| 2 |
from datetime import datetime, timedelta, timezone
|
| 3 |
from typing import Optional
|
| 4 |
|
| 5 |
from fastapi import Depends, HTTPException, status
|
| 6 |
from fastapi.security import OAuth2PasswordBearer
|
| 7 |
from jose import JWTError, jwt
|
| 8 |
-
from passlib.context import CryptContext
|
| 9 |
from cryptography.fernet import Fernet
|
| 10 |
from sqlalchemy.orm import Session
|
| 11 |
|
| 12 |
from .database import get_db
|
| 13 |
from . import models
|
| 14 |
|
|
|
|
| 15 |
SECRET_KEY = os.getenv("SECRET_KEY", "CHANGE_ME_super_secret_key_32bytes!")
|
| 16 |
ALGORITHM = "HS256"
|
| 17 |
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 60 * 24 * 7))
|
| 18 |
|
| 19 |
-
|
| 20 |
-
|
| 21 |
_raw_fernet_key = os.getenv("FERNET_KEY", "")
|
| 22 |
if not _raw_fernet_key:
|
| 23 |
_raw_fernet_key = Fernet.generate_key().decode()
|
|
@@ -28,14 +28,19 @@ fernet = Fernet(_raw_fernet_key.encode() if isinstance(_raw_fernet_key, str) els
|
|
| 28 |
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
|
| 29 |
|
| 30 |
|
| 31 |
-
|
| 32 |
-
|
|
|
|
|
|
|
|
|
|
| 33 |
|
| 34 |
|
| 35 |
-
def
|
| 36 |
-
|
|
|
|
| 37 |
|
| 38 |
|
|
|
|
| 39 |
def encrypt_api_key(api_key: str) -> str:
|
| 40 |
return fernet.encrypt(api_key.encode()).decode()
|
| 41 |
|
|
@@ -44,9 +49,12 @@ def decrypt_api_key(encrypted: str) -> str:
|
|
| 44 |
return fernet.decrypt(encrypted.encode()).decode()
|
| 45 |
|
| 46 |
|
|
|
|
| 47 |
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
| 48 |
to_encode = data.copy()
|
| 49 |
-
expire = datetime.now(timezone.utc) + (
|
|
|
|
|
|
|
| 50 |
to_encode["exp"] = expire
|
| 51 |
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
| 52 |
|
|
@@ -71,4 +79,4 @@ def get_current_user(
|
|
| 71 |
user = db.query(models.User).filter(models.User.username == username).first()
|
| 72 |
if not user:
|
| 73 |
raise exc
|
| 74 |
-
return user
|
|
|
|
| 1 |
import os
|
| 2 |
+
import bcrypt
|
| 3 |
from datetime import datetime, timedelta, timezone
|
| 4 |
from typing import Optional
|
| 5 |
|
| 6 |
from fastapi import Depends, HTTPException, status
|
| 7 |
from fastapi.security import OAuth2PasswordBearer
|
| 8 |
from jose import JWTError, jwt
|
|
|
|
| 9 |
from cryptography.fernet import Fernet
|
| 10 |
from sqlalchemy.orm import Session
|
| 11 |
|
| 12 |
from .database import get_db
|
| 13 |
from . import models
|
| 14 |
|
| 15 |
+
# ββ JWT ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
|
| 16 |
SECRET_KEY = os.getenv("SECRET_KEY", "CHANGE_ME_super_secret_key_32bytes!")
|
| 17 |
ALGORITHM = "HS256"
|
| 18 |
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 60 * 24 * 7))
|
| 19 |
|
| 20 |
+
# ββ Fernet encryption for API keys βββββββββββββββββββββββββββββββββββββββββββββ
|
|
|
|
| 21 |
_raw_fernet_key = os.getenv("FERNET_KEY", "")
|
| 22 |
if not _raw_fernet_key:
|
| 23 |
_raw_fernet_key = Fernet.generate_key().decode()
|
|
|
|
| 28 |
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
|
| 29 |
|
| 30 |
|
| 31 |
+
# ββ Password hashing (pure bcrypt β passlib nahi) ββββββββββββββββββββββββββββββ
|
| 32 |
+
def hash_password(plain: str) -> str:
|
| 33 |
+
# bcrypt max 72 bytes β safely truncate
|
| 34 |
+
secret = plain.encode("utf-8")[:72]
|
| 35 |
+
return bcrypt.hashpw(secret, bcrypt.gensalt()).decode("utf-8")
|
| 36 |
|
| 37 |
|
| 38 |
+
def verify_password(plain: str, hashed: str) -> bool:
|
| 39 |
+
secret = plain.encode("utf-8")[:72]
|
| 40 |
+
return bcrypt.checkpw(secret, hashed.encode("utf-8"))
|
| 41 |
|
| 42 |
|
| 43 |
+
# ββ Fernet helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
|
| 44 |
def encrypt_api_key(api_key: str) -> str:
|
| 45 |
return fernet.encrypt(api_key.encode()).decode()
|
| 46 |
|
|
|
|
| 49 |
return fernet.decrypt(encrypted.encode()).decode()
|
| 50 |
|
| 51 |
|
| 52 |
+
# ββ JWT helpers ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
|
| 53 |
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
| 54 |
to_encode = data.copy()
|
| 55 |
+
expire = datetime.now(timezone.utc) + (
|
| 56 |
+
expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
| 57 |
+
)
|
| 58 |
to_encode["exp"] = expire
|
| 59 |
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
| 60 |
|
|
|
|
| 79 |
user = db.query(models.User).filter(models.User.username == username).first()
|
| 80 |
if not user:
|
| 81 |
raise exc
|
| 82 |
+
return user
|