You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
60 lines
2.1 KiB
60 lines
2.1 KiB
from typing import Optional, Dict, Any |
|
import jwt |
|
from datetime import datetime, timedelta |
|
|
|
class AuthService: |
|
def __init__(self, auth_config: Dict[str, Any]): |
|
self.config = auth_config |
|
self._current_user = None |
|
self._auth_token = None |
|
|
|
def is_authenticated(self) -> bool: |
|
"""Check if user is authenticated""" |
|
return self._auth_token is not None and self._current_user is not None |
|
|
|
def get_current_user(self) -> Optional[Dict[str, Any]]: |
|
"""Get current authenticated user""" |
|
return self._current_user |
|
|
|
def get_auth_token(self) -> Optional[str]: |
|
"""Get current auth token""" |
|
return self._auth_token |
|
|
|
def login(self, username: str, password: str) -> bool: |
|
"""Simple login (in production, this would integrate with external provider)""" |
|
# For demo purposes, simple validation |
|
if username == "admin" and password == "admin": |
|
# Generate a simple JWT token |
|
payload = { |
|
"user_id": "admin", |
|
"username": username, |
|
"role": "admin", |
|
"exp": datetime.utcnow() + timedelta(hours=8) |
|
} |
|
self._auth_token = jwt.encode(payload, self.config['secret_key'], algorithm="HS256") |
|
self._current_user = { |
|
"user_id": "admin", |
|
"username": username, |
|
"role": "admin" |
|
} |
|
return True |
|
return False |
|
|
|
def logout(self): |
|
"""Logout current user""" |
|
self._current_user = None |
|
self._auth_token = None |
|
|
|
def validate_token(self, token: str) -> bool: |
|
"""Validate JWT token""" |
|
try: |
|
payload = jwt.decode(token, self.config['secret_key'], algorithms=["HS256"]) |
|
self._current_user = { |
|
"user_id": payload["user_id"], |
|
"username": payload["username"], |
|
"role": payload["role"] |
|
} |
|
self._auth_token = token |
|
return True |
|
except jwt.InvalidTokenError: |
|
return False
|
|
|