94 lines
3.3 KiB
YAML
94 lines
3.3 KiB
YAML
---
|
||
apiVersion: v1
|
||
kind: ConfigMap
|
||
metadata:
|
||
name: authentication
|
||
namespace: flows
|
||
data:
|
||
authentication.py: |
|
||
import httpx
|
||
|
||
from fastapi import Request, status
|
||
from itsdangerous import URLSafeTimedSerializer, BadData as BadDataError
|
||
from sqladmin.authentication import AuthenticationBackend
|
||
|
||
from flow.config import settings
|
||
|
||
|
||
class AdminAuthentication(AuthenticationBackend):
|
||
def get_serializer(self) -> URLSafeTimedSerializer:
|
||
return URLSafeTimedSerializer(
|
||
secret_key=settings.admin_panel.secret_key,
|
||
)
|
||
|
||
async def login(self, request: Request) -> bool:
|
||
form = await request.form()
|
||
username, password = form["username"], form["password"]
|
||
token_payload = {
|
||
'is_django_user': False,
|
||
'user_id': None,
|
||
}
|
||
# получаем данные о юзере
|
||
if True:
|
||
#print("OK"*30)
|
||
# client: httpx.AsyncClient = settings.django.get_async_session(token=settings.django.token)
|
||
# async with client as session:
|
||
# auth_response = await session.post(url='/login/', json={
|
||
# 'username': username,
|
||
# 'password': password,
|
||
# }, follow_redirects=False)
|
||
# print("*"*30, auth_response.status_code ,settings.django.host, username, password,"*"*30)
|
||
# if auth_response.status_code != status.HTTP_200_OK:
|
||
# return False
|
||
|
||
# django_response = await session.get(url='/client/settings/')
|
||
# print("$"*30, django_response.status_code, "$"*30)
|
||
# if django_response.status_code != status.HTTP_200_OK:
|
||
# return False
|
||
if username == "68Lvh2PsHd5y" and password == "Hnzvh2PsHd5yI1v":
|
||
#user = django_response.json()
|
||
has_access = True
|
||
if not has_access:
|
||
return False
|
||
|
||
user_id = 2
|
||
|
||
token_payload = {
|
||
'is_django_user': True,
|
||
'user_id': user_id,
|
||
}
|
||
else:
|
||
token_payload = {
|
||
'is_django_user': False,
|
||
'user_id': None,
|
||
}
|
||
|
||
token = self.get_serializer().dumps(token_payload)
|
||
|
||
request.session.update({"token": token})
|
||
return True
|
||
|
||
async def logout(self, request: Request) -> bool:
|
||
request.session.clear()
|
||
return True
|
||
|
||
async def authenticate(self, request: Request) -> bool:
|
||
token = request.session.get("token")
|
||
|
||
if not token:
|
||
return False
|
||
|
||
try:
|
||
payload = self.get_serializer().loads(token, max_age=settings.admin_panel.token_max_age)
|
||
except (BadDataError, KeyError):
|
||
return False
|
||
|
||
is_django_user = payload.get('is_django_user')
|
||
if settings.django.use != is_django_user:
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
authentication_backend = AdminAuthentication(secret_key=settings.admin_panel.secret_key)
|