| import json, os, jwt |
| from hashlib import sha256 |
| from fastapi import HTTPException, status, Request |
| from fastapi.security import HTTPBasicCredentials |
| from logger import log_write, logger |
| from datetime import datetime, timedelta |
|
|
|
|
|
|
|
|
| users = json.loads(str(os.getenv("USER_KEYS")).replace("\n", "")) |
| for key in users: |
| if key == "master": continue |
| password = key+users[key]+users["master"] |
| users[key] = sha256(password.encode('UTF-8')).hexdigest() |
|
|
| JWT_SECRET = users["master"] |
| JWT_ALGORITHM = "HS256" |
| JWT_EXPIRATION_TIME_MINUTES = 30 |
|
|
| def authenticate_user(credentials: HTTPBasicCredentials) -> bool: |
| username = credentials.username |
| password = credentials.password |
| password = username+password+users["master"] |
| password = sha256(password.encode('UTF-8')).hexdigest() |
|
|
| if credentials.username not in users or password != users[credentials.username]: |
| log_write(credentials.username, "Autenticacion usuario fallida", "") |
| raise HTTPException( |
| status_code=status.HTTP_401_UNAUTHORIZED, |
| detail="Incorrect username or password", |
| headers={"WWW-Authenticate": "Basic"}, |
| ) |
|
|
| log_write(credentials.username, "Usuario autenticado", "") |
| return True |
|
|
| def create_jwt_token(data): |
| to_encode = {"data": data} |
| expire = datetime.utcnow() + timedelta(minutes=JWT_EXPIRATION_TIME_MINUTES) |
| to_encode.update({"exp": expire}) |
| encoded_jwt = jwt.encode(to_encode, JWT_SECRET, algorithm=JWT_ALGORITHM) |
| return encoded_jwt |
|
|
|
|
|
|
| async def validate_token(request: Request): |
| data = {} |
| try: |
| data = await request.json() |
| token = data.pop("token") |
| payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) |
| data["token_data"] = payload["data"] |
| except Exception as e: |
| logger.error(repr(e) + " - " + str(data)) |
| raise HTTPException(status_code=404, detail="Token inválido") |
| return data |
|
|
|
|
