FastApi production ready backend  for called Vaultexe the password manager
from .auth import reqIpDep
from .cache import AsyncRedisClientDep, SyncRedisClientDep
from .db import dbDep

app/api/deps/__init__.py
Vaultexe/server
from typing import Annotated

from fastapi import Depends, Request
from pydantic import IPvAnyAddress


def get_ip(req: Request) -> IPvAnyAddress:
    """Returns the IP address of the client making the request"""
    if ip := req.headers.get("X-Forwarded-For"):
        return ip.split(",", 1)[0]  # type: ignore
    return req.client.host  # type: ignore


reqIpDep = Annotated[IPvAnyAddress, Depends(get_ip)]

app/api/deps/auth.py
Vaultexe/server
from typing import Annotated

import redis
from fastapi import Depends

from app.cache.client import AsyncRedisClient
from app.core.config import settings


def get_sync_redis_conn() -> redis.Redis:
    return redis.Redis.from_url(str(settings.REDIS_URI))


""" Annotated Dependency """
AsyncRedisClientDep = Annotated[AsyncRedisClient, Depends()]
SyncRedisClientDep = Annotated[redis.Redis, Depends(get_sync_redis_conn)]

app/api/deps/cache.py
Vaultexe/server
from collections.abc import AsyncGenerator
from typing import Annotated

from fastapi import Depends
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import AsyncSessionFactory


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Yields a db session from the pool"""
    async with AsyncSessionFactory() as session:
        session: AsyncSession
        try:
            yield session
        except SQLAlchemyError as e:
            raise e


""" Annotated Dependency """
dbDep = Annotated[AsyncSession, Depends(get_db)]

app/api/deps/db.py
Vaultexe/server
from fastapi import APIRouter, status
from fastapi.responses import Response

from app.api.routes import v1

router = APIRouter()


@router.get("/health")
async def health():
    return Response(status_code=status.HTTP_200_OK)


# Include subrouters
router.include_router(prefix="/v1", router=v1.router, tags=["v1"])

app/api/routes/api.py
Vaultexe/server
from fastapi import APIRouter

from app.api.routes.v1 import device, user

router = APIRouter()

# Include subrouters
router.include_router(prefix="/user", router=user.router, tags=["user"])
router.include_router(prefix="/device", router=device.router, tags=["device"])

app/api/routes/v1/api.py
Vaultexe/server
from collections.abc import AsyncGenerator, Mapping
from contextlib import asynccontextmanager
from typing import Any, overload

import redis.asyncio as aioredis
from redis.asyncio.client import Pipeline

from app.core.config import settings
from app.utils.coders import JsonCoder

async_redis_pool = aioredis.ConnectionPool.from_url(str(settings.REDIS_URI))


class AsyncRedisClient:
    """
    Redis client wrapper.
    """

    def __init__(self) -> None:
        self.redis: aioredis.Redis = aioredis.Redis(connection_pool=async_redis_pool)

    @overload
    async def set(self, key: str, value: Any, *, ttl: int) -> bool:
        ...

    @overload
    async def set(self, key: str, value: Any, *, keepttl: bool) -> bool:
        ...

    async def set(
        self,
        key: str,
        value: Any,
        ttl: int | None = None,
        keepttl: bool | None = None,
    ) -> bool:
        """
        Set a key-value pair in Redis.

        Either ttl or keepttl must be specified but not both.

        Overloads:
            set(self, key: str, value: Any, *, ttl: int) -> bool
            set(self, key: str, value: Any, *, keepttl: bool) -> bool

        Args:
            key (str): Key to set.
            value (Any): Value to set.
            ttl (int, optional): Expiration time in seconds. Defaults to None.
            keepttl (bool, optional): Keep the TTL of the key. Defaults to False.

        Returns:
            bool: True if successful, False otherwise.
        """
        value = self.encode(value)

        if ttl is not None and keepttl is not None:
            raise ValueError("Either ttl or keepttl must be specified but not both.")
        elif ttl is not None:
            res = await self.redis.set(key, value, ex=ttl)
        elif keepttl is not None:
            ttl = await self.redis.ttl(key)
            if ttl == -2:
                # key does not exist
                return False
            res = await self.redis.set(key, value, keepttl=keepttl)
        else:
            raise ValueError("Either ttl or keepttl must be specified but not both.")

        return True if res else False

    async def set_many(self, key_value_pairs: Mapping[str, Any], ttl: int) -> bool:
        """
        Set multiple key-value pairs in Redis.

        Returns:
            True if successful, False otherwise.
        """
        async with self.pipeline() as pipe:
            for key, value in key_value_pairs.items():
                value = self.encode(value)
                pipe.set(key, value, ex=ttl)
            return await pipe.execute()

    async def get(self, key: str) -> Any | None:
        """
        Get a value from Redis.

        Returns:
            Value if successful, None otherwise.
        """
        value = await self.redis.get(key)
        return self.decode(value)

    async def get_with_ttl(self, key: str) -> tuple[int, Any | None]:
        """
        Get a value from Redis with its TTL.

        Returns:
            - (ttl, value) if successful
            - (-2, None) if key does not exist
            - (-1, None) if key has no expire set
        """
        async with self.pipeline() as pipe:
            pipe.ttl(key)
            pipe.get(key)
            ttl, value = await pipe.execute()
        return ttl, self.decode(value)

    async def get_all_startswith(
        self,
        key_prefix: str,
    ) -> dict[str, Any]:
        """
        Get all key-value pairs starting with a given prefix.

        Returns:
            Dictionary of key-value pairs.
        """
        keys = await self.redis.keys(f"{key_prefix}*")
        values = await self.redis.mget(keys)
        values = [self.decode(value) for value in values]
        return dict(zip(keys, values, strict=True))

    async def delete(self, key: str) -> bool:
        """
        Delete a key from Redis.

        Returns:
            True if successful, False otherwise.
        """
        return await self.redis.delete(key) == 1

    async def delete_many(self, keys: list[str]) -> int:
        """
        Delete multiple keys from Redis.

        Returns:
            Number of keys deleted.
        """
        return await self.redis.delete(*keys)

    async def delete_all_startswith(self, key_prefix: str) -> int:
        """
        Delete all keys starting with a given prefix.

        Returns:
            Number of keys deleted.
        """
        keys = await self.get_all_startswith(key_prefix)
        return await self.redis.delete(*keys)

    async def flushall(self) -> bool:
        """Delete all cache"""
        return await self.redis.flushall()

    async def exists(self, key: str) -> bool:
        """Check if a key exists in cache"""
        return await self.redis.exists(key) == 1

    async def expire(self, key: str, expire: int) -> bool:
        """
        Set a key's expiration time in Redis.

        Returns:
            True if successful, False otherwise.
        """
        return await self.redis.expire(key, expire)

    async def ttl(self, key: str) -> int:
        """
        Get a key's TTL in Redis.

        Returns:
            - ttl if key exists
            - -2 if key does not exist
            - -1 if key has no expire set
        """
        return await self.redis.ttl(key)

    @asynccontextmanager
    async def pipeline(self, transactional: bool = True) -> AsyncGenerator[Pipeline, None]:
        """
        Create a pipeline for Redis commands.

        Returns:
            Redis pipeline.
        """
        yield self.redis.pipeline(transaction=transactional)

    def encode(self, value: Any) -> bytes:
        return JsonCoder.encode(value)

    def decode(self, value: bytes | None) -> Any:
        if value is None:
            return None
        return JsonCoder.decode(value)

app/cache/client.py
Vaultexe/server
from enum import Enum


def rt_key_gen(id: str) -> str:
    return f"refresh_token:{id}"


def otp_at_key_gen(id: str) -> str:
    return f"otp_salted_hash:{id}"


def reset_pass_key_gen(token: str) -> str:
    return f"reset_password_token:{token}"


class KeyGen(Enum):
    """
    Enum for the different types of key builders.
    """

    REFRESH_TOKEN = rt_key_gen
    OTP_SALTED_HASH = otp_at_key_gen
    RESET_PASSWORD_TOKEN = reset_pass_key_gen

    def __call__(self, s: str) -> str:
        """Factory method for the key builders"""
        return self.value(s)

app/cache/key_gen.py
Vaultexe/server
from enum import Enum

from pydantic import field_validator
from pydantic_core.core_schema import ValidationInfo
from pydantic_settings import BaseSettings


class Env(str, Enum):
    dev = "dev"
    prod = "prod"


class Settings(BaseSettings):
    """App global settings"""

    # APP
    ENV: Env = Env.dev
    DOMAIN: str
    BACKEND_DOMAIN: str
    PROJECT_NAME: str

    # SECURITY
    OTP_EXPIRE_SECONDS: int
    OTP_LENGTH: int
    JWT_HASHING_ALGORITHM: str = "HS256"
    JWT_SECRET_KEY: str
    REFRESH_TOKEN_EXPIRE_SECONDS: int
    ACCESS_TOKEN_EXPIRE_SECONDS: int

    # DB
    USE_PGBOUNCER: bool = False
    POSTGRES_URI: str | None
    PGBOUNCER_URI: str | None
    DATABASE_DSN: str | None = None

    # Cache
    REDIS_URI: str

    @field_validator("DATABASE_DSN", mode="before")
    def assemble_db_dsn(cls, v, info: ValidationInfo) -> str:
        """Assemble database DSN from environment variables"""
        if info.data["USE_PGBOUNCER"]:
            return info.data["PGBOUNCER_URI"]
        return info.data["POSTGRES_URI"]

    @property
    def is_dev(self) -> bool:
        return self.ENV == Env.dev

    @property
    def is_prod(self) -> bool:
        return self.ENV == Env.prod


settings = Settings()

app/core/config.py
Vaultexe/server
import hashlib
import secrets
import string
from typing import Any

import argon2
import bcrypt
from jose import exceptions as jwt_errors
from jose import jwt as jwt_lib

from app.core.config import settings
from app.utils.exceptions import AuthenticationException, TokenExpiredException


def encode_token(claim: dict[str, Any]) -> str:
    """Encode a JWT token"""
    return jwt_lib.encode(
        claim,
        settings.JWT_SECRET_KEY,
        algorithm=settings.JWT_HASHING_ALGORITHM,
    )


def decode_token(token: str) -> dict[str, Any]:
    """
    Decode a JWT token.

    Auto verifies:
        - The signature
        - The expiration time
        - The audience
        - The issuer

    Returns:
        dict[str, Any]: The decoded JWT token claims

    Raises:
        TokenExpiredException (401): If the token is expired
        AuthenticationError (401): If the token is invalid
    """
    try:
        decoded_token = jwt_lib.decode(
            token,
            settings.JWT_SECRET_KEY,
            algorithms=settings.JWT_HASHING_ALGORITHM,
        )
    except jwt_errors.ExpiredSignatureError:
        raise TokenExpiredException
    except jwt_errors.JWTError:
        raise AuthenticationException

    return decoded_token


def generate_otp() -> str:
    """Generate a otp token."""
    return "".join(secrets.choice(string.digits) for _ in range(settings.OTP_LENGTH))


def get_salted_hash(t: str) -> tuple[str, str]:
    """
    Get a salted hash of a text.

    Returns:
        tuple[str, str]: [salt, hash]
    """
    salt = bcrypt.gensalt()
    hash = sha256_hash(salt + t.encode())
    return salt.decode(), hash


def verify_salted_hash(t: str, salt: str, hash: str) -> bool:
    """
    Verify a salted hash of a text.

    Returns:
        bool: True if the token is valid
    """
    return hash == sha256_hash((salt + t).encode())


def sha256_hash(data: bytes) -> str:
    """Get a sha256 hash of a string"""
    return hashlib.sha256(data, usedforsecurity=False).hexdigest()


def hash_pwd(pwd: str) -> str:
    """
    Hash a password

    Uses argon2id algorithm, it embeds the salt in the hash
    """
    return argon2.PasswordHasher().hash(pwd)


def verify_pwd(pwd: str, hash: str) -> bool:
    """
    Verify a password

    Uses argon2id algorithm, it embeds the salt in the hash
    """
    try:
        argon2.PasswordHasher().verify(hash, pwd)
    except Exception:
        return False
    else:
        return True

app/core/security.py
Vaultexe/server
import datetime as dt
import uuid

from pydantic import IPvAnyAddress

from app.core import security
from app.core.config import settings
from app.schemas import (
    AccessTokenClaim,
    OTPSaltedHash,
    OTPTokenClaim,
    RefreshTokenClaim,
)


def create_web_tokens(
    *,
    subject: uuid.UUID | str,
    ip: IPvAnyAddress,
    is_admin: bool,
    at_ttl: int = settings.ACCESS_TOKEN_EXPIRE_SECONDS,
    rt_ttl: int = settings.REFRESH_TOKEN_EXPIRE_SECONDS,
) -> tuple[AccessTokenClaim, str, RefreshTokenClaim, str]:
    """
    Create [access_token, refresh_token] pair.

    Args:
        subject (uuid.UUID | str): The user's UUID
        ip (IPvAnyAddress): The IP address of the user
        is_admin (bool): Whether the user is an admin
        at_ttl (int, optional): The access token time to live in seconds
        rt_ttl (int, optional): The refresh token time to live in seconds

    Returns:
        tuple[AccessTokenClaim, str, RefreshTokenClaim, str]: [at_claim, at, rt_claim, rt]
    """
    jti, iat = uuid.uuid4(), dt.datetime.now(dt.UTC)

    at_exp = iat + dt.timedelta(seconds=at_ttl)
    rt_exp = iat + dt.timedelta(seconds=rt_ttl)

    at_claim = AccessTokenClaim(jti=jti, sub=subject, is_admin=is_admin, iat=iat, exp=at_exp)
    rt_claim = RefreshTokenClaim(jti=jti, sub=subject, ip=ip, iat=iat, exp=rt_exp)

    at = security.encode_token(at_claim.model_dump())
    rt = security.encode_token(rt_claim.model_dump())

    return at_claim, at, rt_claim, rt


def create_otp_tokens(
    *,
    subject: uuid.UUID | str,
    ip: IPvAnyAddress,
    otp: str,
    otpt_ttl: int = settings.OTP_EXPIRE_SECONDS,
) -> tuple[OTPSaltedHash, OTPTokenClaim, str]:
    """
    Create an OTP token

    It is sent along with the OTP code to the server
    to ensure that the OTP is used from the same device that requested it.
    """
    salt, hash = security.get_salted_hash(otp)

    jti, iat = uuid.uuid4(), dt.datetime.now(dt.UTC)
    exp = iat + dt.timedelta(seconds=otpt_ttl)

    otpt_claim = OTPTokenClaim(jti=jti, sub=subject, ip=ip, iat=iat, exp=exp)
    otp_sh_claim = OTPSaltedHash(jti=jti, sub=subject, ip=ip, iat=iat, exp=exp, salt=salt, hash=hash)

    otpt = security.encode_token(otpt_claim.model_dump())

    return otp_sh_claim, otpt_claim, otpt


def create_reset_pwd_token() -> tuple[str, str]:
    """
    Create a reset password token.

    Returns:
        tuple[str, str]: [reset_token, reset_token_hash]
    """
    reset_token = uuid.uuid4()
    reset_token_hash = security.sha256_hash(reset_token.bytes)
    return reset_token, reset_token_hash

app/core/tokens.py
Vaultexe/server
# isort:skip_file

from .base import BaseRepo

app/db/repos/__init__.py
Vaultexe/server
from abc import ABC
from uuid import UUID

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.base import BaseModel
from app.schemas.base import BaseSchema


class BaseRepo[ModelType: BaseModel, CreateSchemaType: BaseSchema](ABC):
    """
    Base repository class with default CRUD operations
    """

    def __init__(self, model: type[ModelType]) -> None:
        """Initialize repository with model"""
        self.model = model

    async def get(
        self,
        db: AsyncSession,
        *,
        id: UUID | int,
    ) -> ModelType | None:
        """Get model by id"""
        query = sa.select(self.model).filter(self.model.id == id)
        return await db.scalar(query)

    async def create(self, db: AsyncSession, *, obj_in: CreateSchemaType) -> ModelType:
        """
        Create new model.

        Returns:
            Created model
        """
        db_obj = self.model.create_from(obj_in)
        db.add(db_obj)
        return db_obj

    async def bulk_create(self, db: AsyncSession, *, objs_in: list[CreateSchemaType]) -> list[ModelType]:
        """
        Bulk create new models.

        Returns:
            Created models
        """
        db_objs = [self.model.create_from(obj_in) for obj_in in objs_in]
        db.add_all(db_objs)
        return db_objs

app/db/repos/base.py
Vaultexe/server
from app import models, schemas
from app.db.repos.base import BaseRepo


class CipherRepo(BaseRepo[models.Cipher, schemas.CipherCreate]):
    """Cipher repo"""


cipher = CipherRepo(models.Cipher)

app/db/repos/cipher.py
Vaultexe/server
from app import models, schemas
from app.db.repos.base import BaseRepo


class CollectionRepo(BaseRepo[models.Collection, schemas.CollectionCreate]):
    """Cipher repo"""


cipher = CollectionRepo(models.Collection)

app/db/repos/collection.py
Vaultexe/server
from app import models, schemas
from app.db.repos.base import BaseRepo


class UserRepo(BaseRepo[models.User, schemas.UserInvite]):
    """User repo"""


cipher = UserRepo(models.User)

app/db/repos/user.py
Vaultexe/server
"""
Sets up postgresql database connection pool.
"""
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.core.config import settings

async_engine = create_async_engine(
    url=str(settings.DATABASE_DSN),
    echo=settings.is_dev,
    future=True,
    pool_pre_ping=True,
    pool_size=50,  # pgbouncer pool size = 50
    pool_timeout=100,  # pgbouncer pool timeout = 100
)

AsyncSessionFactory = async_sessionmaker(
    async_engine,
    expire_on_commit=False,
)

app/db/session.py
Vaultexe/server
import re
import typing

import sqlalchemy as sa
from sqlalchemy import ClauseElement, Executable, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.compiler import SQLCompiler
from sqlalchemy.sql.selectable import Select

from app.models.base import BaseModel

"""
Inspired by:
    - Fast SQLAlchemy query results exact count
        - https://gist.github.com/hest/8798884
        - https://datawookie.dev/blog/2021/01/sqlalchemy-efficient-counting/
    - Fast SqlAlchemy query approx results count
        - https://gist.github.com/jmcarp/384310cb3925eaa3b3ca
    - Fast & safe SQL approx table row count
        - https://stackoverflow.com/a/7945274/19517403
"""


class ResultCount(typing.NamedTuple):
    """Result of counting operation."""

    count: int
    is_exact: bool


class SQLExplain(Executable, ClauseElement):
    """Explain query execution plan."""

    def __init__(self, stmt: Select, analyze: bool = False):
        self.statement = stmt
        self.analyze = analyze


@compiles(SQLExplain, "postgresql")
def pg_explain(element: SQLExplain, compiler: SQLCompiler, **kw):
    """Compile SQLExplain to SQL."""
    text = "EXPLAIN "
    if element.analyze:
        text += "ANALYZE "
    text += compiler.process(element.statement, **kw)
    return text


def extract_count_from_explain(rows: typing.Sequence) -> int:
    """
    Extract count from Explain output.
    Returns -1 if count cannot be extracted.
    """
    count_pattern = re.compile(r"rows=(\d+)")
    for row in rows:
        match = count_pattern.search(row[0])
        if match:
            return int(match.groups()[0])
    return -1


async def query_results_exact_count(db: AsyncSession, query: Select) -> int:
    """Get exact count of results from a query."""
    counter = query.with_only_columns(func.count(), maintain_column_froms=True)
    res = await db.execute(counter)
    return res.scalar_one()


async def query_results_approx_count(db: AsyncSession, query: Select) -> int:
    """Get approximate count of results from a query."""
    rows = await db.execute(SQLExplain(query))
    rows = rows.all()
    count = extract_count_from_explain(rows)
    return count


async def query_results_count(
    db: AsyncSession,
    query: Select,
    *,
    threshold: int = 10000,
    exact: bool = False,
) -> ResultCount:
    """
    Get count of results from a query.
    If exact is True, exact count is returned.
    If count is less than threshold, exact count is returned.
    """
    if exact:
        count = await query_results_exact_count(db, query)
    else:
        count = await query_results_approx_count(db, query)
        if count < threshold:
            exact = True
            count = await query_results_exact_count(db, query)
    return ResultCount(count, exact)


async def table_approx_count(db: AsyncSession, *, model: type[BaseModel]) -> int:
    """
    Get approximate count of records.

    Reference:
        - https://stackoverflow.com/a/7945274/19517403
    """
    query = sa.text(
        """
        SELECT (
            CASE
                WHEN c.reltuples < 0 THEN 0       -- never vacuumed
                WHEN c.relpages = 0 THEN float8 '0'  -- empty table
                ELSE c.reltuples / c.relpages
            END * (
                pg_catalog.pg_relation_size(c.oid)
                / pg_catalog.current_setting('block_size')::int
            )
        )::bigint
        FROM   pg_catalog.pg_class c
        WHERE  c.oid = :table_name ::regclass;      -- schema-qualified table here
        """
    ).bindparams(table_name=model.table_name())
    result = await db.execute(query)
    return result.scalar_one()


async def table_exact_count(db: AsyncSession, *, model: type[BaseModel]) -> int:
    """Get exact count of records."""
    query = sa.select(sa.func.count(model.id))
    result = await db.execute(query)
    return result.scalar_one()


async def table_count(db: AsyncSession, threshold: int = 10000) -> ResultCount:
    """
    Get exact or approximate count of records.

    If the approximate count < threshold, the exact count is returned.
    """
    count, is_exact = await table_approx_count(db), False
    if count < threshold:
        is_exact = True
        count = await table_exact_count(db)
    return ResultCount(count, is_exact)

app/db/utils.py
Vaultexe/server
"""
    This module is used to check the connection to the database.
    It is called from the Dockerfile as part of the entrypoint.
"""

import asyncio
import logging

from async_timeout import timeout
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import text
from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed

from app.db.session import AsyncSessionFactory

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


TIMEOUT = 5  # seconds
MAX_TRIES = 5
WAIT_TIME = 1  # seconds


@retry(
    stop=stop_after_attempt(MAX_TRIES),
    wait=wait_fixed(WAIT_TIME),
    before=before_log(logger, logging.INFO),
    after=after_log(logger, logging.WARN),
)
async def init_db_connection() -> None:
    """Wait till database connection is established"""
    try:
        async with timeout(TIMEOUT):
            async with AsyncSessionFactory() as session:
                session: AsyncSession
                await session.execute(text("SELECT 1"))
    except Exception as e:
        logger.error("--- Connection to DB failed ---")
        logger.log(logging.ERROR, e)
        raise e


async def main() -> None:
    logger.info("--- Connecting TO POSTGRESQL DB ---")
    await init_db_connection()
    logger.info("--- Connection TO POSTGRESQL DB ESTABLISHED ---")


if __name__ == "__main__":
    asyncio.run(main())

app/entrypoint.py
Vaultexe/server
from fastapi import FastAPI
from fastapi.middleware import cors

from app.api.routes import api
from app.core.config import settings


def add_routers(app: FastAPI) -> None:
    app.include_router(api.router)


def add_middlewares(app: FastAPI) -> None:
    app.add_middleware(
        cors.CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )


def create_app() -> FastAPI:
    """App factory"""
    app = FastAPI(
        title=settings.PROJECT_NAME,
        description="A production grade server example of a password manager",
        openapi_url="/openapi.json",
        docs_url="/docs",
        redoc_url="/redoc",
    )
    add_routers(app)
    add_middlewares(app)
    return app


app = create_app()

app/main.py
Vaultexe/server
"""
Import all models here is essential for alembic to detect them.
"""

# isort: skip_file

from .base import BaseModel

from .user import User

from .device import Device

from .cipher import Cipher

from .collection import Collection

app/models/__init__.py
Vaultexe/server
from functools import lru_cache
from typing import Any, Self

import pydantic
from sqlalchemy import MetaData, inspect
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import Mapper, as_declarative

from app.utils.regex import camel_to_snake

naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}
metadata = MetaData(naming_convention=naming_convention)


@as_declarative(metadata=metadata)
class BaseModel:
    id: Any
    __name__: str

    @declared_attr.directive
    def __tablename__(cls) -> str:
        """Automatically generates a snake case table name"""
        return camel_to_snake(cls.__name__)

    @classmethod
    @lru_cache
    def table_name(cls) -> str:
        """Returns the table name

        If the model is a sub-class of another model in a single table inheritance,
        the table name of the parent model is returned instead
        """
        insp: Mapper = inspect(cls, raiseerr=True)
        return insp.tables[-1].name if insp else cls.__tablename__

    @classmethod
    def create_from(cls, obj: pydantic.BaseModel) -> Self:
        """Creates a new instance of the model from a Pydantic schema instance"""
        model = cls()
        model.import_from(obj)
        return model

    @classmethod
    @lru_cache
    def uq_attrs(cls) -> tuple[str, ...]:
        """Returns a tuple of the model's unique attributes names"""
        insp: Mapper = inspect(cls, raiseerr=True)
        uq_attrs = tuple(attr.name for attr in insp.columns if attr.unique) if insp else ()
        return uq_attrs

    def uq_kwargs(self) -> dict:
        """Returns a dict of the model's unique attributes names and values"""
        return {attr: getattr(self, attr) for attr in self.uq_attrs()}

    def import_from(self, obj: pydantic.BaseModel, **kwargs) -> None:
        """
        Imports values from a Pydantic schema instance to the model instance

        Args:
            obj (BaseSchema): Pydantic schema instance.
            **kwargs: Additional keyword arguments to pass to the Pydantic schema's dict method.

        kwargs:
            include: A list of fields to include in the output.
            exclude: A list of fields to exclude from the output.
            by_alias: Whether to use the field's alias in the dictionary key if defined.
            exclude_unset: Whether to exclude fields that are unset or None from the output.
            exclude_defaults: Whether to exclude fields that are set to their default value from the output.
            exclude_none: Whether to exclude fields that have a value of `None` from the output.
            round_trip: Whether to enable serialization and deserialization round-trip support.
            warnings: Whether to log warnings when invalid fields are encountered.

        Exceptions:
            AttributeError: if schema has an attribute that the model doesn't have
        """
        kwargs.setdefault("by_alias", True)
        d = obj.model_dump(**kwargs)
        for k, v in d.items():
            if hasattr(self, k):
                setattr(self, k, v)
            else:
                raise AttributeError(f"{self.__class__.__name__} has no attribute {k}")

    def __repr__(self) -> str:
        """
        Returns:
            String representation of the model instance

        Examples:
            >>> from app.models import Auction
            >>> auction = Auction(title="test", description="test")
            >>> auction
            <Auction(title='test', description='test')>
        """
        # fmt: off
        columns = ", ".join(
            [
                f"{k}={repr(v)}" for k, v in self.__dict__.items()
                if not k.startswith("_")
            ]
        )
        # fmt: on
        return f"<{self.__class__.__name__}({columns})>"

app/models/base.py
Vaultexe/server
import datetime as dt
import uuid

import sqlalchemy as sa
import sqlalchemy.dialects.postgresql as pg
from sqlalchemy.orm import Mapped, mapped_column

from app.models import BaseModel
from app.models.enums import PgCipherType
from app.schemas.enums import CipherType


class Cipher(BaseModel):
    id: Mapped[uuid.UUID] = mapped_column(pg.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id: Mapped[uuid.UUID] = mapped_column(pg.UUID(as_uuid=True), sa.ForeignKey("user.id"), index=True, nullable=False)
    collection_id: Mapped[uuid.UUID | None] = mapped_column(
        pg.UUID(as_uuid=True), sa.ForeignKey("collection.id"), index=True, nullable=True
    )
    type: Mapped[CipherType] = mapped_column(PgCipherType, nullable=False)
    data: Mapped[bytes] = mapped_column(sa.LargeBinary(), nullable=False)
    created_at: Mapped[dt.datetime] = mapped_column(sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now())
    updated_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime(timezone=True), nullable=True, server_onupdate=sa.func.now())
    deleted_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime(timezone=True), nullable=True)

app/models/cipher.py
Vaultexe/server
import datetime as dt
import uuid

import sqlalchemy as sa
import sqlalchemy.dialects.postgresql as pg
from sqlalchemy.orm import Mapped, mapped_column

from app.models import BaseModel


class Collection(BaseModel):
    id: Mapped[uuid.UUID] = mapped_column(pg.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id: Mapped[uuid.UUID] = mapped_column(pg.UUID(as_uuid=True), sa.ForeignKey("user.id"), index=True, nullable=False)
    name: Mapped[str] = mapped_column(sa.String, nullable=False)
    created_at: Mapped[dt.datetime] = mapped_column(sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now())

app/models/collection.py
Vaultexe/server
import datetime as dt
import uuid

import sqlalchemy.dialects.postgresql as pg
from pydantic import IPvAnyAddress
from sqlalchemy import DateTime, ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.models import BaseModel

"""
max length of ipv6/ipv4 address is 45 characters, see:
https://stackoverflow.com/questions/166132/maximum-length-of-the-textual-representation-of-an-ipv6-address/166157#166157

"""


class Device(BaseModel):
    id: Mapped[uuid.UUID] = mapped_column(pg.UUID(as_uuid=True), primary_key=True, index=True, default=uuid.uuid4)
    user_id: Mapped[uuid.UUID] = mapped_column(pg.UUID(as_uuid=True), ForeignKey("user.id"), index=True, nullable=False)
    user_agent: Mapped[str] = mapped_column(String(length=350), nullable=False)
    regestered_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
    last_login_ip: Mapped[IPvAnyAddress] = mapped_column(String(length=45), nullable=False)
    last_login_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), nullable=False)

app/models/device.py
Vaultexe/server
from sqlalchemy.dialects import postgresql as pg

from app.schemas.enums import CipherType

PgCipherType = pg.ENUM(CipherType, name="cipher_type")

app/models/enums.py
Vaultexe/server
import datetime as dt
import uuid

import sqlalchemy.dialects.postgresql as pg
from sqlalchemy import DateTime, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.models import BaseModel

"""
    Email:
        Max length of email address is 254 characters (We chose max 100), see:
        https://stackoverflow.com/questions/1199190/what-is-the-optimal-length-for-an-email-address-in-a-database
    Password:
        passwords are hashed using agron2 which automatically generates a salt
"""


class User(BaseModel):
    id: Mapped[uuid.UUID] = mapped_column(pg.UUID(as_uuid=True), primary_key=True, index=True, default=uuid.uuid4)
    email: Mapped[str] = mapped_column(String(length=100), nullable=False, unique=True)
    email_verified: Mapped[bool] = mapped_column(nullable=False, default=False)
    is_active: Mapped[bool] = mapped_column(nullable=False, default=False)
    is_admin: Mapped[bool] = mapped_column(nullable=False, default=False)
    master_pwd_hash: Mapped[str] = mapped_column(nullable=False)
    master_pwd_hint: Mapped[str | None] = mapped_column(nullable=True)
    last_pwd_change: Mapped[dt.datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    last_email_change: Mapped[dt.datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())

app/models/user.py
Vaultexe/server
# isort:skip_file

from .base import BaseSchema

from .user import (
    UserInvite,
    UserLogin,
    User,
)

from .device import Device

from .collection import (
    Collection,
    CollectionCreate,
)

from .cipher import (
    Cipher,
    CipherCreate,
)

from .token import (
    TokenBase,
    AccessTokenClaim,
    RefreshTokenClaim,
    WebToken,
    OTPTokenClaim,
    OTPSaltedHash,
)

app/schemas/__init__.py
Vaultexe/server
import pydantic


class BaseSchema(pydantic.BaseModel):
    ...

app/schemas/base.py
Vaultexe/server
import datetime as dt
import uuid

from app.schemas.base import BaseSchema
from app.schemas.enums import CipherType


class Cipher(BaseSchema):
    id: uuid.UUID
    user_id: uuid.UUID
    collection_id: uuid.UUID | None
    type: CipherType
    data: bytes
    created_at: dt.datetime
    updated_at: dt.datetime | None
    deleted_at: dt.datetime | None


class CipherCreate(BaseSchema):
    user_id: uuid.UUID
    collection_id: uuid.UUID | None
    type: CipherType
    data: bytes


class CipherUpdate(BaseSchema):
    collection_id: uuid.UUID | None
    data: bytes | None

app/schemas/cipher.py
Vaultexe/server
import datetime as dt
import uuid

from app.schemas.base import BaseSchema


class Collection(BaseSchema):
    id: uuid.UUID
    user_id: uuid.UUID
    name: str
    created_at: dt.datetime


class CollectionCreate(BaseSchema):
    user_id: uuid.UUID
    name: str

app/schemas/collection.py
Vaultexe/server
import datetime as dt
import uuid

from app.schemas.base import BaseSchema


class Device(BaseSchema):
    id: uuid.UUID
    user_id: uuid.UUID
    user_agent: str
    regeistered_at: dt.datetime
    last_login_ip: dt.datetime
    last_login_at: dt.datetime


class DeviceCreate(BaseSchema):
    id: uuid.UUID
    user_id: uuid.UUID
    user_agent: str


def user_agent_extract(ua: str) -> dict:
    """Extracts device id, device type and os from user agent string."""

app/schemas/device.py
Vaultexe/server
from enum import Enum, auto

from typing_extensions import override


class BaseEnum(str, Enum):
    """
    Base class for all enums
    """

    @override
    def _generate_next_value_(name, start, count, last_values) -> str:  # type: ignore
        """
        Overrides the enum function that generates the next value
        Uses the name as the automatic value, rather than an integer
        See https://docs.python.org/3/library/enum.html#using-automatic-values for reference
        """
        return name

    @classmethod
    def as_dict(cls) -> dict:
        return {e.name: e.value for e in cls}


class CipherType(BaseEnum):
    LOGIN = auto()
    NOTE = auto()


class TokenType(BaseEnum):
    ACCESS = auto()
    REFRESH = auto()
    OTP = auto()
    RESET_PASSWORD = auto()

app/schemas/enums.py
Vaultexe/server
import datetime as dt
import uuid

from pydantic import Field, IPvAnyAddress

from app.schemas import BaseSchema
from app.schemas.enums import TokenType

"""
JWT: JWT Token Schema

JWT:
    jti: JWT ID
    iat: Issued At
    exp: Expires At
    sub: Subject
    ip: IP Address
"""


class TokenBase(BaseSchema):
    type: TokenType
    jti: uuid.UUID
    iat: dt.datetime
    exp: dt.datetime
    sub: uuid.UUID | str
    is_admin: bool | None = Field(None, exclude=True)
    ip: IPvAnyAddress | None = Field(None, exclude=True)


class AccessTokenClaim(TokenBase):
    type: TokenType = TokenType.ACCESS
    is_admin: bool


class RefreshTokenClaim(TokenBase):
    type: TokenType = TokenType.REFRESH
    ip: IPvAnyAddress


class WebToken(BaseSchema):
    access_token: str
    expires_in: int
    token_type: str = "bearer"
    refresh_token: str
    refresh_token_expires_in: int


class OTPTokenClaim(TokenBase):
    type: TokenType = TokenType.OTP
    ip: IPvAnyAddress


class OTPSaltedHash(OTPTokenClaim):
    salt: str
    hash: str


class OTPWebToken(BaseSchema):
    otp_access_token: str
    expires_in: int
    token_type: str = "bearer"


class ResetPasswordTokenClaim(TokenBase):
    type: TokenType = TokenType.RESET_PASSWORD
    ip: IPvAnyAddress

app/schemas/token.py
Vaultexe/server
import datetime as dt
import uuid
from typing import Self

from pydantic import EmailStr, computed_field, model_validator

from app.core.security import hash_pwd
from app.schemas.base import BaseSchema
from app.utils.regex import generate_password


class UserInvite(BaseSchema):
    email: EmailStr

    @computed_field(description="Auto generated password on invitation")
    @property
    def master_pwd_hash() -> str:
        pwd = generate_password()
        return hash_pwd(pwd)


class UserLogin(BaseSchema):
    email: EmailStr
    master_pwd: str


class UserResetPassword(BaseSchema):
    curr_pwd_hash: str
    new_pwd_hash: str

    @model_validator(mode="after")
    def check_pwds_diff(self) -> Self:
        """Check that new password is different from current password."""
        if self.curr_pwd_hash and self.new_pwd_hash and self.curr_pwd_hash == self.new_pwd_hash:
            raise ValueError("New password must be different from current password")
        return self


class User(BaseSchema):
    id: uuid.UUID
    email: EmailStr
    email_verified: bool
    last_pwd_change: dt.datetime | None
    last_email_change: dt.datetime | None

app/schemas/user.py
Vaultexe/server
import datetime as dt
import json
from typing import Any

from fastapi.encoders import jsonable_encoder
from starlette.responses import JSONResponse

"""
Heavily inspired by fastapi-cache
https://github.com/long2ice/fastapi-cache/blob/ee58f979d4ef22a03012d86efb6fdb7b5f5b91c6/fastapi_cache/coder.py#L53
"""

TYPE_CONVERTER = {
    "datetime": lambda x: dt.datetime.fromisoformat(x),
    "date": lambda x: dt.date.fromisoformat(x),
}


class JsonEncoder(json.JSONEncoder):
    """
    Custom JSON encoder to handle custom types over JSON.
    """

    def default(self, obj: Any) -> Any:
        """
        Override json.JSONEncoder.default method to handle custom types.
        Convert certain custom types to their JSON representation.
        """
        if isinstance(obj, dt.datetime):
            return {"__type__": "datetime", "val": obj.isoformat()}
        elif isinstance(obj, dt.date):
            return {"__type__": "date", "val": obj.isoformat()}
        return jsonable_encoder(obj)


def object_hook(obj: dict) -> Any:
    """Decode back custom types"""
    type = obj.get("__type__", None)

    if type is None:
        return obj

    if type not in TYPE_CONVERTER:
        raise TypeError(f"Type {type} not supported")

    return TYPE_CONVERTER[type](obj["val"])


class JsonCoder:
    """
    Json coder for encoding and decoding values.
    Uses custom JSON encoder & object hook decoder
    """

    @classmethod
    def encode(cls, value: Any) -> bytes:
        if isinstance(value, JSONResponse):
            return value.body
        return json.dumps(value, cls=JsonEncoder).encode()

    @classmethod
    def decode(cls, value: bytes) -> Any:
        return json.loads(value, object_hook=object_hook)

app/utils/coders.py
Vaultexe/server
from fastapi import HTTPException, status


class TokenExpiredException(HTTPException):
    def __init__(
        self,
        detail: str = "Token has expired",
        headers: dict[str, str] | None = None,
    ) -> None:
        headers = headers or {"WWW-Authenticate": "Bearer"}
        super().__init__(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=detail,
            headers=headers,
        )


class AuthenticationException(TokenExpiredException):
    def __init__(self) -> None:
        super().__init__(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication failed",
        )

app/utils/exceptions.py
Vaultexe/server
import re
import secrets
import string


def camel_to_snake(text: str) -> str:
    """
    Converts CamelCase to snake_case

    Usage:
        >>> camel_to_snake('HTTPResponseCodeXYZ')
        'http_response_code_xyz'
        >>> camel_to_snake('getHTTPResponseCode')
        'get_http_response_code'
    """
    # https://stackoverflow.com/questions/1175208/elegant-python-function-to-convert-camelcase-to-snake-case
    expression = r"((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))"
    replacement_exp = r"_\1"
    return re.sub(expression, replacement_exp, text).lower()


_rpunctuation = string.punctuation + "\\\\"
rchars = string.ascii_letters + string.digits + _rpunctuation + " "


def generate_password(min_length: int = 16) -> str:
    """
    Generates a random password.

    Args:
        min_length: minimum length of password (default: 16)

    Constraints:
        At least 8 characters long
        At least 1 uppercase letter
        At least 1 lowercase letter
        At least 1 digit
        At least 1 special character

    Usage:
        >>> generate_password()
        'wq5Y0!h6#X3@9^E7'
        >>> generate_password(8)
        '9^E7wq5Y'
    """
    if min_length < 8:
        raise ValueError("Minimum length of password should be at least 8 characters.")

    c1 = secrets.choice(string.ascii_lowercase)
    c2 = secrets.choice(string.ascii_uppercase)
    c3 = secrets.choice(string.digits)
    c4 = secrets.choice(string.punctuation)

    password = "".join(secrets.choice(rchars) for _ in range(min_length - 4))
    return password + c1 + c2 + c3 + c4


def is_valid_password(password: str) -> bool:
    """
    Checks if password is valid.

    Valid password:
        - At least 8 characters long
        - At least 1 uppercase letter
        - At least 1 lowercase letter
        - At least 1 digit
        - At least 1 special character

    Usage:
        >>> is_valid_password('Password123!')
        True
        >>> is_valid_password('password123!')
        False

    References:
        - https://stackoverflow.com/questions/19605150/regex-for-password-must-contain-at-least-eight-characters-at-least-one-number-a

    """
    expression = (
        rf"^(?=.*[{string.ascii_uppercase}])"  # At least 1 uppercase letter
        rf"(?=.*[{string.ascii_lowercase}])"  # At least 1 lowercase letter
        rf"(?=.*[{string.digits}])"  # At least 1 digit
        rf"(?=.*[{_rpunctuation}])"  # At least 1 special character
        f"[{rchars}]"
        r"{8,}$"  # At least 8 characters long
    )
    match = re.match(expression, password)
    return bool(match)

app/utils/regex.py
Vaultexe/server
import asyncio
from logging.config import fileConfig

from alembic import context
from alembic.operations.ops import MigrationScript
from alembic.script import ScriptDirectory
from sqlalchemy import engine_from_config, pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import AsyncEngine

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

""" Custom settings for alembic. """
# Sets alembic.ini sqlalchemy.url variable to environment setting DATABASE_URI
from app.core.config import settings  # noqa

config.set_main_option("sqlalchemy.url", str(settings.DATABASE_DSN))

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
# Used custom metadata for alembic autogenerate support
""" Custom settings """
from app.models.base import metadata as base_model_metadata  # noqa

target_metadata = base_model_metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def process_revision_directives(context: context, _ , directives: list[MigrationScript]):
    """
    Custom revision directive to automatically generate revision id's sequentially.
    Reference: https://stackoverflow.com/a/67398484/19517403
    """
    # extract Migration
    migration_script = directives[0]
    # extract current head revision
    head_revision = ScriptDirectory.from_config(context.config).get_current_head()

    if head_revision is None:
        # edge case with first migration
        new_rev_id = 1
    else:
        # default branch with incrementation
        last_rev_id = int(head_revision.lstrip('0'))
        new_rev_id = last_rev_id + 1
    # fill zeros up to 4 digits: 1 -> 0001
    migration_script.rev_id = f'{new_rev_id:04}'


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        process_revision_directives=process_revision_directives,
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        process_revision_directives=process_revision_directives,
    )

    with context.begin_transaction():
        context.run_migrations()


async def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    connectable = AsyncEngine(
        engine_from_config(
            config.get_section(config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
            future=True,
        )
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())

migration/env.py
Vaultexe/server
"""create_enums

Revision ID: cf9857c61471
Revises:
Create Date: 2023-06-21 20:14:40.886402

"""
from alembic import op

from app.models.enums import PgCipherType

# revision identifiers, used by Alembic.
revision = '0001'
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    """
    Enums are created automatically when they are used in a table.
    This is a workaround to create them manually.
    """
    pass

def downgrade() -> None:
    """
    Drop enums:
        - PgCipherType
    """
    PgCipherType.drop(op.get_bind(), checkfirst=True)

migration/versions/0001_create_enums.py
Vaultexe/server
"""create_user

Revision ID: 0001
Revises:
Create Date: 2023-09-27 20:40:37.639208

"""
import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision = '0002'
down_revision = '0001'
branch_labels = None
depends_on = None


def upgrade() -> None:
    """ Create user table """
    op.create_table(
        'user',
        sa.Column('id', sa.UUID(), nullable=False),
        sa.Column('email', sa.String(length=100), nullable=False),
        sa.Column('email_verified', sa.Boolean(), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False),
        sa.Column('is_admin', sa.Boolean(), nullable=False),
        sa.Column('master_pwd_hash', sa.String(), nullable=False),
        sa.Column('master_pwd_hint', sa.String(), nullable=True),
        sa.Column('last_pwd_change', sa.DateTime(timezone=True), nullable=True),
        sa.Column('last_email_change', sa.DateTime(timezone=True), nullable=True),
        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
        sa.PrimaryKeyConstraint('id', name=op.f('pk_user')),
        sa.UniqueConstraint('email', name=op.f('uq_user_email')),
    )
    op.create_index(op.f('ix_user_id'), 'user', ['id'], unique=False)


def downgrade() -> None:
    """ Drop user table """
    op.drop_index(op.f('ix_user_id'), table_name='user')
    op.drop_table('user')

migration/versions/0002_create_user.py
Vaultexe/server
"""create_device

Revision ID: 0002
Revises: 0001
Create Date: 2023-09-29 00:51:51.741909

"""
import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision = "0003"
down_revision = "0002"
branch_labels = None
depends_on = None


def upgrade() -> None:
    """ Create device table """
    op.create_table(
        "device",
        sa.Column("id", sa.UUID(), nullable=False),
        sa.Column("user_id", sa.UUID(), nullable=False),
        sa.Column("user_agent", sa.String(length=350), nullable=False),
        sa.Column("regestered_at",sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
        sa.Column("last_login_ip", sa.String(length=45), nullable=False),
        sa.Column("last_login_at", sa.DateTime(timezone=True), nullable=False),
        sa.ForeignKeyConstraint(["user_id"], ["user.id"], name=op.f("fk_device_user_id_user")),
        sa.PrimaryKeyConstraint("id", name=op.f("pk_device")),
    )
    op.create_index(op.f("ix_device_id"), "device", ["id"], unique=False)
    op.create_index(op.f("ix_device_user_id"), "device", ["user_id"], unique=False)


def downgrade() -> None:
    """ Drop device table """
    op.drop_index(op.f("ix_device_user_id"), table_name="device")
    op.drop_index(op.f("ix_device_id"), table_name="device")
    op.drop_table("device")

migration/versions/0003_create_device.py
Vaultexe/server
"""create_cipher

Revision ID: 0004
Revises: 0003
Create Date: 2023-10-03 22:32:44.951197

"""
import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision = "0004"
down_revision = "0003"
branch_labels = None
depends_on = None


def upgrade() -> None:
    """Create collection table"""
    op.create_table(
        "collection",
        sa.Column("id", sa.UUID(), nullable=False),
        sa.Column("user_id", sa.UUID(), nullable=False),
        sa.Column("name", sa.String(), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
        sa.ForeignKeyConstraint(["user_id"], ["user.id"], name=op.f("fk_collection_user_id_user")),
        sa.PrimaryKeyConstraint("id", name=op.f("pk_collection")),
    )
    op.create_index(op.f("ix_collection_user_id"), "collection", ["user_id"], unique=False)


def downgrade() -> None:
    """Drop collection table table"""
    op.drop_index(op.f("ix_collection_user_id"), table_name="collection")
    op.drop_table("collection")

migration/versions/0004_create_collection.py
Vaultexe/server
"""create_cipher

Revision ID: 0005
Revises: 0004
Create Date: 2023-10-03 22:51:12.404481

"""
import sqlalchemy as sa
from alembic import op

from app.models.enums import PgCipherType

# revision identifiers, used by Alembic.
revision = "0005"
down_revision = "0004"
branch_labels = None
depends_on = None


def upgrade() -> None:
    """ Create cipher table """
    op.create_table(
        "cipher",
        sa.Column("id", sa.UUID(), nullable=False),
        sa.Column("user_id", sa.UUID(), nullable=False),
        sa.Column("collection_id", sa.UUID(), nullable=True),
        sa.Column("type", PgCipherType, nullable=False),
        sa.Column("data", sa.LargeBinary(), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.ForeignKeyConstraint(["collection_id"], ["collection.id"], name=op.f("fk_cipher_collection_id_collection")),
        sa.ForeignKeyConstraint(["user_id"], ["user.id"], name=op.f("fk_cipher_user_id_user")),
        sa.PrimaryKeyConstraint("id", name=op.f("pk_cipher")),
    )
    op.create_index(op.f("ix_cipher_user_id"), "cipher", ["user_id"], unique=False)
    op.create_index(op.f("ix_cipher_collection_id"), "cipher", ["collection_id"], unique=False)

def downgrade() -> None:
    """ Drop cipher table """
    op.drop_index(op.f("ix_cipher_user_id"), table_name="cipher")
    op.drop_index(op.f("ix_cipher_collection_id"), table_name="cipher")
    op.drop_table("cipher")

migration/versions/0005_create_cipher.py
Vaultexe/server
import pytest

from app.utils.regex import camel_to_snake, generate_password, is_valid_password


@pytest.mark.parametrize('input, expected', [
    ('', ''),
    ('HTTP', 'http'),
    ('CamelCase', 'camel_case'),
    ('camel_camel_case', 'camel_camel_case'),
    ('HTTPResponseCode', 'http_response_code'),
    ('HTTPResponseCodeXYZ', 'http_response_code_xyz'),
])
def test_camel_to_snake(input: str, expected: str):
    assert camel_to_snake(input) == expected


@pytest.mark.parametrize('input, expected', [
    ('', False),
    ('12345678', False),
    ('password', False),
    ('Password', False),
    ('Password123', False),
    ('!@#$%^&*()_+-', False),
    ('Password123!', True),
    ('Password123[', True),
    ('Aa1!@#$%^&*()_+-', True),
])
def test_is_valid_password(input: str, expected: bool):
    assert is_valid_password(input) == expected


@pytest.mark.parametrize('input, expected', [
    (generate_password(), True)
    for _ in range(100)
])
def test_generate_password(input: str, expected: bool):
    assert is_valid_password(input) == expected

tests/utils/test_regex.py
Vaultexe/server