mirror of
https://github.com/Noettore/lagomareGateKeeperBot.git
synced 2025-10-15 03:26:40 +02:00
229 lines
7.5 KiB
Python
229 lines
7.5 KiB
Python
import json
|
|
from enum import Enum
|
|
from datetime import datetime, timezone
|
|
from commons import Status, Credential
|
|
|
|
class Role(Enum):
|
|
ADMIN = "admin"
|
|
MEMBER = "member"
|
|
GUEST = "guest"
|
|
|
|
class _Grant:
|
|
def __init__(self, grantor: str, expires_at: datetime, granted_at: datetime = datetime.now(), last_used_at: datetime = None, status: Status = Status.ENABLED):
|
|
self._grantor: str = grantor
|
|
self._granted_at: datetime = granted_at
|
|
self._expires_at = expires_at
|
|
self._last_used_at = last_used_at
|
|
self._status: Status = status
|
|
|
|
def __dict__(self):
|
|
return {
|
|
"grantor": self._grantor,
|
|
"granted_at": self._granted_at.isoformat(),
|
|
"expires_at": self._expires_at.isoformat(),
|
|
"last_used_at": self._last_used_at.isoformat() if self._last_used_at else None,
|
|
"status": self._status.value
|
|
}
|
|
|
|
@property
|
|
def grantor(self) -> str:
|
|
return self._grantor
|
|
|
|
@grantor.setter
|
|
def grantor(self, grantor: str):
|
|
self._grantor = grantor
|
|
|
|
@property
|
|
def granted_at(self) -> datetime:
|
|
return self._granted_at
|
|
|
|
@property
|
|
def expires_at(self) -> datetime:
|
|
return self._expires_at
|
|
|
|
@property
|
|
def last_used_at(self) -> datetime:
|
|
return self._last_used_at
|
|
|
|
@property
|
|
def status(self) -> Status:
|
|
return self._status
|
|
|
|
class _User:
|
|
def __init__(self, id: str, username: str, fullname: str, role: Role = Role.GUEST, credentials: Credential = Credential("", ""), grants: dict = None, status: Status = Status.ENABLED):
|
|
self._id: str = id
|
|
self._username: str = username
|
|
self._fullname: str = fullname
|
|
self._role: Role = role if isinstance(role, Role) else Role(role)
|
|
self._credentials: Credential = credentials
|
|
self._grants: dict[str, _Grant] = {gate:_Grant(grant.get("grantor", ""), datetime.fromisoformat(grant.get("expires_at").replace("Z", "+00:00"))) for gate, grant in grants.items()} if grants else {}
|
|
self._status: Status = status
|
|
|
|
def __dict__(self):
|
|
return {
|
|
"id": self._id,
|
|
"username": self._username,
|
|
"fullname": self._fullname,
|
|
"role": self._role.value,
|
|
"credentials": self._credentials.__dict__(),
|
|
"grants": {gate: grant.__dict__() for gate, grant in self._grants.items()},
|
|
"status": self._status.value
|
|
}
|
|
|
|
@property
|
|
def id(self) -> str:
|
|
return self._id
|
|
|
|
@property
|
|
def username(self) -> str:
|
|
return self._username
|
|
|
|
@username.setter
|
|
def username(self, username: str):
|
|
self._username = username
|
|
|
|
@property
|
|
def fullname(self) -> str:
|
|
return self._fullname
|
|
|
|
@fullname.setter
|
|
def fullname(self, fullname: str):
|
|
self._fullname = fullname
|
|
|
|
@property
|
|
def role(self) -> Role:
|
|
return self._role
|
|
|
|
@role.setter
|
|
def role(self, role: str | Role):
|
|
self._role = role if isinstance(role, Role) else Role(role)
|
|
|
|
@property
|
|
def credentials(self) -> Credential:
|
|
return self._credentials
|
|
|
|
@credentials.setter
|
|
def credentials(self, credentials: Credential):
|
|
self._credentials = credentials
|
|
|
|
@property
|
|
def grants(self) -> dict[str, _Grant]:
|
|
return self._grants
|
|
|
|
@property
|
|
def status(self) -> Status:
|
|
return self._status
|
|
|
|
@status.setter
|
|
def status(self, status: int | Status):
|
|
self._status = status if isinstance(status, Status) else Status(status)
|
|
|
|
def grant(self, gate: str, grant: _Grant):
|
|
self._grants[gate] = grant
|
|
|
|
def revoke(self, gate: str):
|
|
self._grants.pop(gate, None)
|
|
|
|
|
|
class Users:
|
|
def __init__(self, json_path: str = "./data/users.json"):
|
|
self._json_path: str = json_path
|
|
self._users = self._load_users()
|
|
|
|
def _load_users(self) -> dict[str, _User]:
|
|
try:
|
|
with open(self._json_path, "r") as f:
|
|
users = {}
|
|
for uid, info in json.load(f).items():
|
|
try:
|
|
user = _User(
|
|
id=uid,
|
|
username=info.get("username", ""),
|
|
fullname=info.get("fullname", ""),
|
|
role=info.get("role", Role.GUEST),
|
|
credentials=Credential(info.get("credentials", {}).get("username", ""), info.get("credentials", {}).get("password", "")),
|
|
grants={gate: info for gate, info in info.get("grants", {}).items()},
|
|
status=info.get("status", Status.ENABLED)
|
|
)
|
|
users[uid] = user
|
|
except:
|
|
continue
|
|
return users
|
|
except Exception:
|
|
return {}
|
|
|
|
def _save_users(self) -> None:
|
|
with open(self._json_path, "w") as f:
|
|
json.dump({uid: user.__dict__ for uid, user in self._users.items()}, f, default=str)
|
|
|
|
def update_user(self, id: str, username: str, fullname: str) -> bool:
|
|
if id in self._users.keys():
|
|
self._users[id].username = username
|
|
self._users[id].fullname = fullname
|
|
else:
|
|
self._users[id] = _User(id, username, fullname)
|
|
self._save_users()
|
|
return True
|
|
|
|
def get_username(self, id: str) -> str:
|
|
if id in self._users.keys():
|
|
return self._users[id].username
|
|
return None
|
|
|
|
def get_fullname(self, id: str) -> str:
|
|
if id in self._users.keys():
|
|
return self._users[id].fullname
|
|
return None
|
|
|
|
def get_status(self, id: str) -> Status:
|
|
if id in self._users.keys():
|
|
return self._users[id].status
|
|
return Status.DISABLED
|
|
|
|
def get_role(self, id: str) -> Role:
|
|
if id in self._users.keys():
|
|
return self._users[id].role
|
|
return Role.GUEST
|
|
|
|
def get_credentials(self, id: str) -> Credential:
|
|
if id in self._users.keys():
|
|
return self._users[id].credentials
|
|
return None
|
|
|
|
def set_credentials(self, id: str, credentials: Credential) -> bool:
|
|
if id in self._users.keys():
|
|
self._users[id].credentials = credentials
|
|
return True
|
|
return False
|
|
|
|
def can_open_gate(self, id: str, gate: str) -> bool:
|
|
if id in self._users.keys():
|
|
if gate in self._users[id].grants.keys():
|
|
if self._users[id].grants[gate].status == Status.ENABLED:
|
|
if self._users[id].grants[gate].expires_at > datetime.now(timezone.utc):
|
|
return True
|
|
return False
|
|
|
|
def get_grantor(self, id: str, gate: str) -> str:
|
|
if id in self._users.keys():
|
|
if gate in self._users[id].grants.keys():
|
|
return self._users[id].grants[gate].grantor
|
|
return None
|
|
|
|
def get_admins(self) -> list[str]:
|
|
return [uid for uid, user in self._users.items() if user.role == Role.ADMIN]
|
|
|
|
def grant_access(self, id: str, gate: str, expires_at: datetime, grantor_id: str) -> bool:
|
|
if id in self._users.keys():
|
|
self._users[id].grant(gate, _Grant(grantor_id, expires_at))
|
|
self._save_users()
|
|
return True
|
|
return False
|
|
|
|
def revoke_access(self, id: str, gate: str) -> bool:
|
|
if id in self._users.keys():
|
|
self._users[id].revoke(gate)
|
|
self._save_users()
|
|
return True
|
|
return False
|