Merge tokens

Author: zale-mateo

Backend/copia

@@ -0,0 +1,11 @@
+ENV="DEV"
+ROOT_PATH_DEV=""
+ROOT_PATH_PROD="/api"
+DB_URL="sqlite:///./src/mi-db-sqlite.db"
+MQTT_TOPIC="test_topic"
+MQTT_HOST="localhost"
+MQTT_PORT=1883
+MQTT_KEEPALIVE=60
+
+SECRET_KEY="your_secret_key"
+ALGORITHM="HS256"

Backend/src/main.py

@@ -85,6 +85,13 @@ def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
 @app.get("/users/profile")
 def profile(my_user: Annotated[dict, Depends(decode_token)]):
     return my_user
+'''#PARA HACER LA REDIRECCION
+@app.get("/protected-route")
+def check_auth():
+    raise HTTPException(
+        status_code=401,
+        detail="Usuario no autenticado. Redirigir al login."
+    )'''
 
 
 def mi_callback(mensaje: str) -> None:

Backend/src/nodo/auth/__init__.py

@@ -0,0 +1,15 @@
+# src/nodo/auth/crud.py
+from sqlalchemy.orm import Session
+from src.nodo import models
+from . import  hashing
+
+def get_user_by_username(db: Session, username: str):
+    return db.query(models.Registro).filter(models.Registro.username == username).first()
+
+def create_user(db: Session, username: str, password: str):
+    hashed_password = hashing.hash_password(password)
+    db_user = models.Registro(username=username, password=hashed_password)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user

Backend/src/nodo/auth/crud.py

@@ -0,0 +1,10 @@
+# src/nodo/auth/hashing.py
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+def hash_password(password: str):
+    return pwd_context.hash(password)
+
+def verify_password(plain_password: str, hashed_password: str):
+    return pwd_context.verify(plain_password, hashed_password)

Backend/src/nodo/auth/hashing.py

@@ -0,0 +1,117 @@
+# src/nodo/auth/jwt.py
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+from src.config_db import get_db
+from src.nodo.auth.crud import get_user_by_username
+from fastapi import Depends, HTTPException, status, Request
+from sqlalchemy.orm import Session
+from fastapi.security import OAuth2PasswordBearer
+
+SECRET_KEY = "mi_clave_secreta"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+def create_access_token(data: dict):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+'''def verify_token(token: str):
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        return payload if "exp" in payload and payload["exp"] >= datetime.utcnow().timestamp() else None
+    except JWTError:
+        return None
+    
+def get_cookie_token(request: Request):
+    token = request.cookies.get("access_token")
+    print (request.cookies)
+    if not token:
+        print("No se encontró token en las cookies, utilizando uno de prueba...")
+        token = "<tu_token_de_prueba>"
+    return token
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="iniciar_sesion")
+def get_current_user(db: Session = Depends(get_db), token: str = Depends(get_cookie_token)):
+    try:
+        print(token)
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="No se pudo validar el usuario",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        # Obtener el usuario de la base de datos
+        user = get_user_by_username(db, username=username)
+        if user is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Usuario no encontrado",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        return user
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token inválido o expirado",
+            headers={"WWW-Authenticate": "Bearer"},
+        )'''
+
+
+def verify_token(token: str):
+    """
+    Verifica el token JWT y retorna el payload si es válido.
+    """
+    try:
+        # Configura para verificar automáticamente la expiración del token
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM], options={"verify_exp": True})
+        return payload
+    except JWTError:
+        return None
+
+def get_cookie_token(request: Request):
+    token = request.cookies.get("access_token")
+    print('TOKEN:')
+    print(token)
+    if not token:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="No se encontró un token de acceso en las cookies.",
+        )
+    if token.startswith("Bearer "):
+        token = token[len("Bearer "):]
+    print("Token limpio:", token)  # Debug
+    return token
+
+def get_current_user(db: Session = Depends(get_db), token: str = Depends(get_cookie_token)):
+    """
+    Valida el token JWT y retorna el usuario actual.
+    """
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="No se pudo validar el usuario",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        # Obtener el usuario de la base de datos
+        user = get_user_by_username(db, username=username)
+        if user is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Usuario no encontrado",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        return user
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token inválido o expirado",
+            headers={"WWW-Authenticate": "Bearer"},
+        )

Backend/src/nodo/auth/jwt.py

@@ -57,8 +57,8 @@ class Medicion(BaseModel):
 class Registro(BaseModel):
     __tablename__ = 'registros'
 
-    usuario = Column(String(10), primary_key=True)
-    contrasenia = Column(String(8), nullable=False)
+    username = Column(String(10), primary_key=True)
+    password = Column(String(8), nullable=False)
 
 # Modelo para Alerta
 class Alerta(BaseModel):