kiwi-vpn/api/kiwi_vpn_api/db/user.py

256 lines
5.8 KiB
Python

"""
Python representation of `user` table.
"""
from __future__ import annotations
from typing import Any, Sequence
from pydantic import root_validator
from sqlalchemy.exc import IntegrityError
from sqlmodel import Field, Relationship, SQLModel
from ..config import Config
from .connection import Connection
from .device import Device
from .tag import Tag, TagValue
class UserBase(SQLModel):
"""
Common to all representations of users
"""
name: str = Field(primary_key=True)
email: str | None = Field(default=None)
country: str | None = Field(default=None)
state: str | None = Field(default=None)
city: str | None = Field(default=None)
organization: str | None = Field(default=None)
organizational_unit: str | None = Field(default=None)
class UserCreate(UserBase):
"""
Representation of a newly created user
"""
password: str | None = Field(default=None)
password_clear: str | None = Field(default=None)
@root_validator
@classmethod
def hash_password(cls, values: dict[str, Any]) -> dict[str, Any]:
"""
Ensure the `password` value of this user gets set.
"""
if (values.get("password")) is not None:
# password is set
return values
if (password_clear := values.get("password_clear")) is None:
raise ValueError("No password to hash")
if (current_config := Config._) is None:
raise ValueError("Not configured")
values["password"] = current_config.crypto.crypt_context.hash(
password_clear)
return values
class UserRead(UserBase):
"""
Representation of a user read via the API
"""
pass
class User(UserBase, table=True):
"""
Representation of `user` table
"""
password: str
tags: list[Tag] = Relationship(
back_populates="user",
sa_relationship_kwargs={
"lazy": "joined",
"cascade": "all, delete-orphan",
},
)
devices: list[Device] = Relationship(
back_populates="owner",
)
@classmethod
def create(
cls,
*,
user: UserCreate,
) -> User | None:
"""
Create a new user in the database.
"""
try:
with Connection.session as db:
new_user = cls.from_orm(user)
db.add(new_user)
db.commit()
db.refresh(new_user)
return new_user
except IntegrityError:
# user already existed
return None
@classmethod
def get(cls, name: str) -> User | None:
"""
Load user from database by name.
"""
with Connection.session as db:
return db.get(cls, name)
@classmethod
def authenticate(
cls,
name: str,
password: str,
) -> User | None:
"""
Authenticate with name/password against users in database.
"""
crypt_context = Config._.crypto.crypt_context
if (user := cls.get(name)) is None:
# nonexistent user, fake doing password verification
crypt_context.dummy_verify()
return None
if not crypt_context.verify(password, user.password):
# password hash mismatch
return None
if TagValue.login in user.get_tags():
# no login permission
return None
return user
def update(self) -> None:
"""
Update this user in the database.
"""
with Connection.session as db:
db.add(self)
db.commit()
db.refresh(self)
def delete(self) -> None:
"""
Delete this user from the database.
"""
with Connection.session as db:
db.delete(self)
db.commit()
def get_tags(self) -> set[TagValue]:
"""
Return the tags of this user.
"""
return set(
tag._
for tag in self.tags
)
def set_tags(
self,
tags: Sequence[TagValue],
) -> None:
"""
Change the tags of this user.
"""
self.tags = [
Tag(
user_name=self.name,
tag_value=tag.value,
) for tag in tags
]
def may_edit(
self,
target: User | Device,
) -> bool:
"""
Check if this user can edit another user or a device.
"""
# admin can "edit" everything
if TagValue.admin in self.get_tags():
return True
# user can "edit" itself
if isinstance(target, User) and target != self:
return False
# user can edit its owned devices
return target.owner == self
def may_admin(
self,
target: User | Device,
) -> bool:
"""
Check if this user can administer another user or a device.
"""
# only admin can "admin" anything
if TagValue.admin not in self.get_tags():
return False
# admin canot "admin itself"!
if isinstance(target, User) and target == self:
return False
# admin can "admin" everything else
return True
def may_create(
self,
target: type,
owner: User | None = None,
) -> bool:
"""
Check if this user can create another user or a device.
"""
# can never create anything but users or devices
if not issubclass(target, (User, Device)):
return False
# admin can "create" everything
if TagValue.admin in self.get_tags():
return True
# user can only create devices for itself
if target is Device and owner == self:
return True
# deny be default
return False