lenticular_cloud2/lenticular_cloud/model.py

258 lines
8.4 KiB
Python
Raw Normal View History

2020-05-09 18:00:07 +00:00
from flask import current_app
from flask_login import UserMixin
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
2020-05-10 12:34:28 +00:00
from collections.abc import MutableSequence
from datetime import datetime
from dateutil import tz
2020-05-10 12:34:28 +00:00
import pyotp
import json
import logging
import crypt
2022-07-15 08:53:06 +00:00
import secrets
import string
2023-10-09 19:58:44 +00:00
from sqlalchemy import null
2023-09-30 10:58:24 +00:00
from sqlalchemy.orm import DeclarativeBase, MappedAsDataclass, Mapped, mapped_column, relationship, declarative_base
2023-03-18 11:00:40 +00:00
from flask_sqlalchemy import SQLAlchemy
2023-09-30 10:58:24 +00:00
from flask_sqlalchemy.model import Model, DefaultMeta
from flask_sqlalchemy.extension import _FSAModel
2022-02-20 15:54:35 +00:00
from flask_migrate import Migrate
from datetime import datetime
import uuid
2023-10-20 08:05:30 +00:00
from typing import Iterator, Optional, List, Dict, Tuple, Any, Type, TYPE_CHECKING
2022-06-17 11:38:49 +00:00
from cryptography.x509 import Certificate as CertificateObj
2022-06-18 11:05:18 +00:00
from sqlalchemy.ext.declarative import DeclarativeMeta
2020-05-09 18:00:07 +00:00
logger = logging.getLogger(__name__)
2020-05-09 18:00:07 +00:00
2020-05-09 18:00:07 +00:00
2022-07-15 08:53:06 +00:00
db = SQLAlchemy()
2022-06-17 11:38:49 +00:00
migrate = Migrate()
2023-09-30 10:58:24 +00:00
class BaseModelIntern(MappedAsDataclass, DeclarativeBase):
pass
if TYPE_CHECKING:
class BaseModel (_FSAModel,BaseModelIntern):
pass
else:
BaseModel: Type[_FSAModel] = db.Model
2023-10-09 19:58:44 +00:00
2023-09-30 10:58:24 +00:00
class ModelUpdatedMixin:
2023-10-09 19:58:44 +00:00
created_at: Mapped[datetime] = mapped_column(db.DateTime, default=datetime.now(), nullable=False)
modified_at: Mapped[datetime] = mapped_column(db.DateTime, default=datetime.now(), onupdate=datetime.now, nullable=False)
2022-06-18 11:05:18 +00:00
2020-05-09 18:00:07 +00:00
class SecurityUser(UserMixin):
def __init__(self, username):
self._username = username
def get_id(self):
return self._username
class Service(object):
2022-06-18 11:05:18 +00:00
def __init__(self, name: str):
2020-05-09 18:00:07 +00:00
self._name = name
2022-07-15 08:53:06 +00:00
self._app_token = False
2023-12-25 23:24:28 +00:00
self._icon: Optional[str] = None
self._href: Optional[str] = None
2020-05-09 18:00:07 +00:00
self._client_cert = False
self._pki_config = {
'cn': '{username}',
'email': '{username}@{domain}'
}
@staticmethod
2022-06-18 17:35:05 +00:00
def from_config(name, config) -> 'Service':
2020-05-09 18:00:07 +00:00
service = Service(name)
2022-07-15 08:53:06 +00:00
if 'app_token' in config:
service._app_token = bool(config['app_token'])
2020-05-09 18:00:07 +00:00
if 'client_cert' in config:
service._client_cert = bool(config['client_cert'])
if 'pki_config' in config:
2020-05-10 17:37:57 +00:00
service._pki_config.update(config['pki_config'])
2023-12-25 23:24:28 +00:00
if 'icon' in config:
service._icon = str(config['icon'])
if 'href' in config:
service._href = str(config['href'])
2020-05-09 18:00:07 +00:00
return service
@property
2022-06-18 11:05:18 +00:00
def name(self) -> str:
2020-05-09 18:00:07 +00:00
return self._name
@property
2022-06-18 11:05:18 +00:00
def client_cert(self) -> bool:
2020-05-09 18:00:07 +00:00
return self._client_cert
2022-07-15 08:53:06 +00:00
@property
def app_token(self) -> bool:
return self._app_token
2020-05-09 18:00:07 +00:00
@property
2022-06-18 11:05:18 +00:00
def pki_config(self) -> dict[str,str]:
2020-05-09 18:00:07 +00:00
if not self._client_cert:
raise Exception('invalid call')
return self._pki_config
2023-12-25 23:24:28 +00:00
@property
def icon(self) -> Optional[str]:
return self._icon
@property
def href(self) -> Optional[str]:
return self._href
2020-05-09 18:00:07 +00:00
class Certificate(object):
2022-06-18 11:05:18 +00:00
def __init__(self, cn: str, ca_name: str, cert_data: CertificateObj, revoked=False):
2020-05-09 18:00:07 +00:00
self._cn = cn
self._ca_name = ca_name
self._cert_data = cert_data
self._revoked = revoked
self._cert_data.not_valid_after.replace(tzinfo=tz.tzutc())
self._cert_data.not_valid_before.replace(tzinfo=tz.tzutc())
2020-05-09 18:00:07 +00:00
@property
2022-06-18 11:05:18 +00:00
def cn(self) -> str:
2020-05-09 18:00:07 +00:00
return self._cn
@property
2022-06-18 11:05:18 +00:00
def ca_name(self) -> str:
2020-05-09 18:00:07 +00:00
return self._ca_name
@property
def not_valid_before(self) -> datetime:
return self._cert_data.not_valid_before.replace(tzinfo=tz.tzutc()).astimezone(tz.tzlocal()).replace(tzinfo=None)
2020-05-09 18:00:07 +00:00
@property
def not_valid_after(self) -> datetime:
return self._cert_data.not_valid_after.replace(tzinfo=tz.tzutc()).astimezone(tz.tzlocal()).replace(tzinfo=None)
2020-05-09 18:00:07 +00:00
@property
def serial_number(self) -> int:
return self._cert_data.serial_number
@property
def serial_number_hex(self) -> str:
return f'{self._cert_data.serial_number:X}'
def fingerprint(self, algorithm=hashes.SHA256()) -> bytes:
2020-05-09 18:00:07 +00:00
return self._cert_data.fingerprint(algorithm)
@property
def is_valid(self) -> bool:
return self.not_valid_after > datetime.now() and not self._revoked
def pem(self) -> str:
2020-05-09 18:00:07 +00:00
return self._cert_data.public_bytes(encoding=serialization.Encoding.PEM).decode()
@property
def raw(self):
return self._cert_data
2020-05-09 18:00:07 +00:00
def __str__(self):
return f'Certificate(cn={self._cn}, ca_name={self._ca_name}, not_valid_before={self.not_valid_before}, not_valid_after={self.not_valid_after})'
2020-05-10 12:34:28 +00:00
def generate_uuid():
return str(uuid.uuid4())
2020-05-10 12:34:28 +00:00
2023-10-09 19:58:44 +00:00
class User(BaseModel, ModelUpdatedMixin):
id: Mapped[uuid.UUID] = mapped_column(db.Uuid, primary_key=True, default=uuid.uuid4)
username: Mapped[str] = mapped_column(db.String, unique=True, nullable=False)
password_hashed: Mapped[str] = mapped_column(db.String, nullable=False)
alternative_email: Mapped[Optional[str]] = mapped_column( db.String, nullable=True)
last_login: Mapped[Optional[datetime]] = mapped_column(db.DateTime, nullable=True)
enabled: Mapped[bool] = mapped_column(db.Boolean, nullable=False, default=False)
app_tokens: Mapped[List['AppToken']] = relationship('AppToken', back_populates='user')
2023-12-25 16:28:09 +00:00
passkey_credentials: Mapped[List['PasskeyCredential']] = relationship('PasskeyCredential', back_populates='user', cascade='delete,delete-orphan', passive_deletes=True)
2023-10-09 19:58:44 +00:00
def __init__(self, **kwargs) -> None:
2022-06-18 11:05:18 +00:00
super().__init__(**kwargs)
2020-05-10 12:34:28 +00:00
2020-05-09 18:00:07 +00:00
@property
2022-06-18 11:05:18 +00:00
def is_authenticated(self) -> bool:
2020-05-09 18:00:07 +00:00
return True # TODO
2022-06-18 11:05:18 +00:00
def get(self, key) -> None:
2022-02-06 22:57:01 +00:00
print(f'getitem: {key}') # TODO
2020-05-09 18:00:07 +00:00
@property
2022-07-15 08:53:06 +00:00
def groups(self) -> list['Group']:
2023-12-24 10:09:41 +00:00
admins = current_app.config['ADMINS']
if self.username in admins:
return [Group(name='admin')]
else:
return []
2020-05-09 18:00:07 +00:00
@property
2022-02-06 22:57:01 +00:00
def email(self) -> str:
2020-05-30 21:43:55 +00:00
domain = current_app.config['DOMAIN']
return f'{self.username}@{domain}'
2020-05-09 18:00:07 +00:00
2022-07-15 08:53:06 +00:00
def change_password(self, password_new: str) -> None:
self.password_hashed = crypt.crypt(password_new)
2023-10-22 17:45:37 +00:00
def get_token_by_name(self, name: str) -> Optional['AppToken']:
for token in self.app_tokens:
if token.name == name:
return token
return None
2022-07-15 08:53:06 +00:00
2023-10-20 08:05:30 +00:00
def get_token_by_scope(self, scope: str) -> Iterator['AppToken']:
2022-07-15 08:53:06 +00:00
for token in self.app_tokens:
2023-10-20 08:05:30 +00:00
if scope in token.scopes.split():
yield token # type: ignore
2022-07-15 08:53:06 +00:00
2020-05-09 18:00:07 +00:00
2023-10-09 19:58:44 +00:00
class AppToken(BaseModel, ModelUpdatedMixin):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
2023-10-20 08:05:30 +00:00
scopes: Mapped[str] = mapped_column(nullable=False) # string of a list seperated by `,`
2023-10-09 19:58:44 +00:00
user_id: Mapped[uuid.UUID] = mapped_column(
db.Uuid,
2022-07-15 08:53:06 +00:00
db.ForeignKey(User.id), nullable=False)
2023-10-09 19:58:44 +00:00
user: Mapped[User] = relationship(User, back_populates="app_tokens")
token: Mapped[str] = mapped_column(nullable=False)
name: Mapped[str] = mapped_column(nullable=False)
last_used: Mapped[Optional[datetime]] = mapped_column(db.DateTime, nullable=True, default=None)
2020-05-09 18:00:07 +00:00
2022-07-15 08:53:06 +00:00
@staticmethod
2023-10-20 08:05:30 +00:00
def new(user: User, scopes: str, name: str):
2022-07-15 08:53:06 +00:00
alphabet = string.ascii_letters + string.digits
2023-10-09 19:58:44 +00:00
token = ''.join(secrets.choice(alphabet) for i in range(12))
2023-10-20 08:05:30 +00:00
return AppToken(scopes=scopes, token=token, user=user, name=name)
2023-10-09 19:58:44 +00:00
2022-04-08 19:28:22 +00:00
2023-12-25 16:28:09 +00:00
class PasskeyCredential(BaseModel, ModelUpdatedMixin): # pylint: disable=too-few-public-methods
"""Passkey credential model"""
2022-04-08 19:28:22 +00:00
2023-10-09 19:58:44 +00:00
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
user_id: Mapped[uuid.UUID] = mapped_column(db.Uuid, db.ForeignKey('user.id', ondelete='CASCADE'), nullable=False)
credential_id: Mapped[bytes] = mapped_column(db.LargeBinary, nullable=False)
credential_public_key: Mapped[bytes] = mapped_column(db.LargeBinary, nullable=False)
2023-10-09 19:58:44 +00:00
name: Mapped[str] = mapped_column(db.String(250), nullable=False)
2023-12-25 16:28:09 +00:00
last_used: Mapped[Optional[datetime]] = mapped_column(db.DateTime, nullable=True, default=None)
sign_count: Mapped[int] = mapped_column(db.Integer, nullable=False, default=0)
2022-04-08 19:28:22 +00:00
2023-12-25 16:28:09 +00:00
user = db.relationship('User', back_populates='passkey_credentials')
2022-04-08 19:28:22 +00:00
2023-10-09 19:58:44 +00:00
class Group(BaseModel, ModelUpdatedMixin):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(db.String(), nullable=False, unique=True)