User Auth Done

backend/app/core/security.py

@@ -0,0 +1,61 @@
+import os
+import base64
+from datetime import datetime, timedelta
+from typing import Optional
+from jose import JWTError, jwt
+import bcrypt
+from Crypto.PublicKey import RSA
+from Crypto.Cipher import PKCS1_OAEP
+from fastapi import HTTPException
+
+from app.core.config import SECRET_KEY, ALGORITHM
+
+RSA_PRIVATE_KEY_PATH = "private.pem"
+RSA_PUBLIC_KEY_PATH = "public.pem"
+
+if not os.path.exists(RSA_PRIVATE_KEY_PATH) or not os.path.exists(RSA_PUBLIC_KEY_PATH):
+    key = RSA.generate(2048)
+    private_key = key.export_key()
+    with open(RSA_PRIVATE_KEY_PATH, "wb") as file_out:
+        file_out.write(private_key)
+
+    public_key = key.publickey().export_key()
+    with open(RSA_PUBLIC_KEY_PATH, "wb") as file_out:
+        file_out.write(public_key)
+else:
+    with open(RSA_PRIVATE_KEY_PATH, "rb") as f:
+        private_key = f.read()
+    with open(RSA_PUBLIC_KEY_PATH, "rb") as f:
+        public_key = f.read()
+
+rsa_private_key = RSA.import_key(private_key)
+cipher_rsa = PKCS1_OAEP.new(rsa_private_key)
+
+def decrypt_password(encrypted_b64_password: str) -> str:
+    try:
+        encrypted_bytes = base64.b64decode(encrypted_b64_password)
+        decrypted = cipher_rsa.decrypt(encrypted_bytes)
+        return decrypted.decode("utf-8")
+    except Exception as e:
+        raise HTTPException(status_code=400, detail="Invalid RSA payload or decryption failed")
+
+def verify_password(plain_password, hashed_password):
+    if isinstance(plain_password, str):
+        plain_password = plain_password.encode('utf-8')
+    if isinstance(hashed_password, str):
+        hashed_password = hashed_password.encode('utf-8')
+    return bcrypt.checkpw(plain_password, hashed_password)
+
+def get_password_hash(password):
+    if isinstance(password, str):
+        password = password.encode('utf-8')
+    return bcrypt.hashpw(password, bcrypt.gensalt()).decode('utf-8')
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)