basyx commited on
Commit
87cb704
·
verified ·
1 Parent(s): 8f7138e

Update auth/security.py

Browse files
Files changed (1) hide show
  1. auth/security.py +64 -49
auth/security.py CHANGED
@@ -3,120 +3,135 @@ from datetime import datetime, timedelta, timezone
3
  from typing import Optional, Dict, Any
4
 
5
  from jose import jwt, JWTError
 
6
  from fastapi import HTTPException, status, Depends
7
  from fastapi.security import OAuth2PasswordBearer
8
 
9
- from passlib.context import CryptContext
10
-
11
 
12
  # =========================================================
13
- # CONFIG (ENV-FIRST SECURITY MODEL)
14
  # =========================================================
15
 
16
  SECRET_KEY = os.getenv("SECRET_KEY")
17
- ALGORITHM = os.getenv("ALGORITHM", "HS256")
18
- ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "1440"))
19
 
20
  if not SECRET_KEY:
21
- raise RuntimeError("SECRET_KEY is not set in environment variables")
 
 
 
 
 
 
22
 
23
 
24
  # =========================================================
25
- # PASSWORD HASHING (FIXED - NO BCRYPT)
 
 
 
 
 
 
 
26
  # =========================================================
27
- # Why this change:
28
- # - bcrypt is broken in your runtime (passlib backend + __about__ bug)
29
- # - bcrypt has 72-byte limitation causing signup crashes
30
- # - argon2 is modern, memory-hard, and production safe
31
 
32
  pwd_context = CryptContext(
33
- schemes=["argon2"],
34
  deprecated="auto",
35
- argon2__time_cost=2,
36
- argon2__memory_cost=102400,
37
- argon2__parallelism=8,
38
  )
39
 
40
 
41
  def hash_password(password: str) -> str:
42
- if not isinstance(password, str) or not password.strip():
43
- raise ValueError("Invalid password input")
 
 
 
 
44
 
45
  return pwd_context.hash(password)
46
 
47
 
48
- def verify_password(plain_password: str, hashed_password: str) -> bool:
49
- if not plain_password or not hashed_password:
50
- return False
 
51
 
52
  try:
53
- return pwd_context.verify(plain_password, hashed_password)
 
 
 
54
  except Exception:
55
  return False
56
 
57
 
58
  # =========================================================
59
- # JWT HANDLING
60
  # =========================================================
61
 
62
  def create_access_token(
63
  data: Dict[str, Any],
64
- expires_delta: Optional[timedelta] = None
65
  ) -> str:
66
 
67
  to_encode = data.copy()
68
 
69
  expire = datetime.now(timezone.utc) + (
70
- expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
 
 
71
  )
72
 
73
  to_encode.update({"exp": expire})
74
 
75
- return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 
 
 
 
 
76
 
 
77
 
78
- def decode_token(token: str) -> Dict[str, Any]:
79
  try:
80
- return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
 
 
 
 
 
81
  except JWTError:
 
82
  raise HTTPException(
83
  status_code=status.HTTP_401_UNAUTHORIZED,
84
- detail="Invalid or expired authentication token",
85
  headers={"WWW-Authenticate": "Bearer"},
86
  )
87
 
88
 
89
  # =========================================================
90
- # FASTAPI AUTH DEPENDENCY
91
  # =========================================================
92
 
93
- oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
 
 
 
94
 
 
 
 
95
 
96
- def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict[str, Any]:
97
  payload = decode_token(token)
98
 
99
  user_id = payload.get("sub")
100
 
101
  if not user_id:
 
102
  raise HTTPException(
103
- status_code=status.HTTP_401_UNAUTHORIZED,
104
- detail="Invalid authentication payload",
105
  )
106
 
107
- return payload
108
-
109
-
110
- # =========================================================
111
- # PASSWORD POLICY (HARDENED)
112
- # =========================================================
113
-
114
- def validate_password_strength(password: str) -> None:
115
- if len(password) < 8:
116
- raise ValueError("Password must be at least 8 characters")
117
-
118
- if password.isdigit():
119
- raise ValueError("Password cannot be only numeric")
120
-
121
- if password.lower() in {"password", "12345678", "qwerty"}:
122
- raise ValueError("Password is too weak")
 
3
  from typing import Optional, Dict, Any
4
 
5
  from jose import jwt, JWTError
6
+ from passlib.context import CryptContext
7
  from fastapi import HTTPException, status, Depends
8
  from fastapi.security import OAuth2PasswordBearer
9
 
 
 
10
 
11
  # =========================================================
12
+ # ENV CONFIG
13
  # =========================================================
14
 
15
  SECRET_KEY = os.getenv("SECRET_KEY")
 
 
16
 
17
  if not SECRET_KEY:
18
+ raise RuntimeError("SECRET_KEY environment variable missing")
19
+
20
+ ALGORITHM = os.getenv("ALGORITHM", "HS256")
21
+
22
+ ACCESS_TOKEN_EXPIRE_MINUTES = int(
23
+ os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "1440")
24
+ )
25
 
26
 
27
  # =========================================================
28
+ # PASSWORD HASHING
29
+ # =========================================================
30
+ # IMPORTANT:
31
+ # pbkdf2_sha256 avoids:
32
+ # - bcrypt crashes
33
+ # - bcrypt native dependency issues
34
+ # - 72-byte limits
35
+ # - passlib backend bugs
36
  # =========================================================
 
 
 
 
37
 
38
  pwd_context = CryptContext(
39
+ schemes=["pbkdf2_sha256"],
40
  deprecated="auto",
 
 
 
41
  )
42
 
43
 
44
  def hash_password(password: str) -> str:
45
+
46
+ if not password:
47
+ raise ValueError("Password required")
48
+
49
+ if len(password) < 8:
50
+ raise ValueError("Password too short")
51
 
52
  return pwd_context.hash(password)
53
 
54
 
55
+ def verify_password(
56
+ plain_password: str,
57
+ hashed_password: str,
58
+ ) -> bool:
59
 
60
  try:
61
+ return pwd_context.verify(
62
+ plain_password,
63
+ hashed_password,
64
+ )
65
  except Exception:
66
  return False
67
 
68
 
69
  # =========================================================
70
+ # JWT
71
  # =========================================================
72
 
73
  def create_access_token(
74
  data: Dict[str, Any],
75
+ expires_delta: Optional[timedelta] = None,
76
  ) -> str:
77
 
78
  to_encode = data.copy()
79
 
80
  expire = datetime.now(timezone.utc) + (
81
+ expires_delta
82
+ if expires_delta
83
+ else timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
84
  )
85
 
86
  to_encode.update({"exp": expire})
87
 
88
+ return jwt.encode(
89
+ to_encode,
90
+ SECRET_KEY,
91
+ algorithm=ALGORITHM,
92
+ )
93
+
94
 
95
+ def decode_token(token: str):
96
 
 
97
  try:
98
+ return jwt.decode(
99
+ token,
100
+ SECRET_KEY,
101
+ algorithms=[ALGORITHM],
102
+ )
103
+
104
  except JWTError:
105
+
106
  raise HTTPException(
107
  status_code=status.HTTP_401_UNAUTHORIZED,
108
+ detail="Invalid token",
109
  headers={"WWW-Authenticate": "Bearer"},
110
  )
111
 
112
 
113
  # =========================================================
114
+ # AUTH DEPENDENCY
115
  # =========================================================
116
 
117
+ oauth2_scheme = OAuth2PasswordBearer(
118
+ tokenUrl="/api/auth/login"
119
+ )
120
+
121
 
122
+ def get_current_user(
123
+ token: str = Depends(oauth2_scheme)
124
+ ):
125
 
 
126
  payload = decode_token(token)
127
 
128
  user_id = payload.get("sub")
129
 
130
  if not user_id:
131
+
132
  raise HTTPException(
133
+ status_code=401,
134
+ detail="Invalid authentication",
135
  )
136
 
137
+ return payload