digimint
113ebce9e1
- SVG icons are now preprocessed into raw HTML at first request. This allows them to be inlined (use `{{icon(icon_name)|safe}}`) and thus styled by CSS.
- General CSS improvements (especially around buttons)
- A basic role editor is now implemented. Go to `/namespace/<id>/role` to see it.
- Task and invite lists now have an "add new" button on the list page.
- Slight permission fixes
- Added `assert_left()` and `assert_right()` to `Either`s. Now, if you do `if isinstance(x, Right)`, you can `x.assert_left()` in the `else` to make the type checker happy.
267 lines
No EOL
8.7 KiB
Python
267 lines
No EOL
8.7 KiB
Python
from dataclasses import Field, dataclass, field
|
|
import dataclasses
|
|
from enum import Enum, auto
|
|
import logging
|
|
import os
|
|
from typing import Any, Self, get_args, get_origin, override
|
|
|
|
from taskflower.types.either import Either, Left, Right, gather_results
|
|
from taskflower.types.option import Nothing, Option, Some
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
class ConfigFormatError(Exception):
|
|
pass
|
|
|
|
class ConfigKeyError[T](Exception):
|
|
@override
|
|
def __init__(
|
|
self,
|
|
field: Field[T],
|
|
exc: Exception
|
|
):
|
|
self.field: Field[T] = field
|
|
self.exc: Exception = exc
|
|
|
|
super().__init__(str(self))
|
|
|
|
@override
|
|
def __str__(self) -> str:
|
|
return f'[Config key `{self.field.name}`] {self.__class__.__name__}: {str(self.exc)}'
|
|
|
|
@override
|
|
def __repr__(self) -> str:
|
|
return f'{self.__class__.__name__}({repr(self.field)}, {repr(self.exc)})'
|
|
|
|
class ConfigKeyMissingError[T](ConfigKeyError[T]):
|
|
@override
|
|
def __init__(self, field: Field[T]):
|
|
super().__init__(field, KeyError(f'Configuration key {field.name} not found in configuration source!'))
|
|
|
|
class ConfigKeyInvalidError[T](ConfigKeyError[T]):
|
|
pass
|
|
|
|
@dataclass(frozen=True)
|
|
class FieldVal[T]:
|
|
key: str
|
|
val: T
|
|
|
|
class EnumFromEnv(Enum):
|
|
@classmethod
|
|
def from_env(cls, env_val: str) -> Either[Exception, Self]:
|
|
for k, v in cls._member_map_.items():
|
|
if k.upper() == env_val.upper():
|
|
if isinstance(v, cls):
|
|
return Right(v)
|
|
else:
|
|
return Left(ValueError(f'{cls.__name__}._member_map_ value is not an instance of {cls.__name__}!'))
|
|
|
|
return Left(KeyError(f'No such key `{env_val}` in enum {cls.__name__}'))
|
|
|
|
class SignUpMode(EnumFromEnv):
|
|
''' Restrictions on who can sign up for an account.
|
|
|
|
OPEN: Anyone can sign up for an account, no registration code required.
|
|
USERS_CAN_INVITE: An invite code is required to create an account. Any
|
|
user can generate an invite code.
|
|
ADMINS_CAN_INVITE: An invite code is requird to create an account. Only
|
|
system administrators can generate an invite code.
|
|
'''
|
|
OPEN = auto()
|
|
USERS_CAN_INVITE = auto()
|
|
ADMINS_CAN_INVITE = auto()
|
|
|
|
class HIBPMode(EnumFromEnv):
|
|
''' Whether to download a local copy of the HaveIBeenPwned API. Note that
|
|
the database is very large (about 40GB at time of writing).
|
|
|
|
LOCAL_ONLY: Download the database and ONLY use the downloaded copy. With
|
|
this option set, password hashes will never be sent to the
|
|
API.
|
|
HYBRID: Download the database and keep it as a backup in case the API
|
|
goes offline. With this option set, password hashes will be sent
|
|
to the API whenever possible, but if the API goes down, the site
|
|
will still be able to check passwords.
|
|
ONLINE_ONLY: Do NOT download the database. With this option set,
|
|
password hashes will ALWAYS be sent to the API. If the API
|
|
goes down, breachlist checks will be bypassed entirely.
|
|
'''
|
|
LOCAL_ONLY = auto()
|
|
HYBRID = auto()
|
|
ONLINE_ONLY = auto()
|
|
|
|
class HIBPLocalCacheMode(EnumFromEnv):
|
|
''' Whether to keep the local API copy in the database or store it as local
|
|
files. Note that there are literal billions of rows in the data, and as
|
|
such, the database size will be much, much larger than the local file
|
|
size (more than 256GB in my testing).
|
|
|
|
STORE_IN_DB: Store the local cache data in the database. This allows
|
|
for faster lookup times, but requires a very large amount
|
|
of space in the database.
|
|
STORE_AS_FILES: Store the local cache as files in the local filesystem.
|
|
This is a bit slower, but requires far less space.
|
|
'''
|
|
STORE_IN_DB = auto()
|
|
STORE_AS_FILES = auto()
|
|
|
|
@dataclass(frozen=True)
|
|
class ConfigType:
|
|
# Application secrets
|
|
db_secret : str # Secret value used to 'pepper' password hashes. This MUST
|
|
# be generated randomly and cryptographically securely.
|
|
# For an example of how to do this:
|
|
# https://flask.palletsprojects.com/en/stable/config/#SECRET_KEY
|
|
#
|
|
# In a multi-node environment, this key must be the same
|
|
# across all nodes.
|
|
|
|
app_secret : str # Secret value used to generate session tokens. This should
|
|
# ALSO be generated securely, and it should be different from
|
|
# ``db_secret``
|
|
|
|
# URL to submit issues to
|
|
issue_url: Option[str] = field(default_factory=Nothing[str])
|
|
|
|
# Users
|
|
sign_up_mode: SignUpMode = SignUpMode.ADMINS_CAN_INVITE
|
|
|
|
# Whether to keep a local copy of the HaveIBeenPwned API
|
|
hibp_mode: HIBPMode = HIBPMode.ONLINE_ONLY
|
|
|
|
# How to keep the local HIBP API copy
|
|
hibp_local_mode: HIBPLocalCacheMode = HIBPLocalCacheMode.STORE_AS_FILES
|
|
hibp_local_dir : Option[str] = field(default_factory=Nothing[str])
|
|
|
|
# Database connection URL
|
|
db_url: str = 'sqlite:///site.db'
|
|
|
|
# Data directory path
|
|
data_path: str = 'instance'
|
|
|
|
# Debug settings
|
|
debug: bool = False
|
|
|
|
# Regenerate icon file with each request. Useful during development, but it
|
|
# should never be enabled in production.
|
|
debug_always_regen_icon_file: bool = False
|
|
|
|
|
|
@classmethod
|
|
def from_env(cls) -> Either[list[ConfigKeyError[Any]], Self]: # pyright:ignore[reportExplicitAny]
|
|
def _try_get_field[T](fld: Field[T]) -> Either[ConfigKeyError[T], FieldVal[T]]:
|
|
def _try_as_type(raw: str, as_t: type[T]) -> Either[Exception, T]:
|
|
origin, args = get_origin(as_t), get_args(as_t)
|
|
|
|
check_t = (
|
|
origin if origin
|
|
else as_t
|
|
)
|
|
|
|
if not isinstance(check_t, type):
|
|
return Left(TypeError(f'Configuration key `{fld.name}` uses unsupported type {str(check_t)}!')) # pyright:ignore[reportAny]
|
|
elif check_t is str:
|
|
return Right[Exception, T](raw) # pyright:ignore[reportArgumentType]
|
|
elif check_t is int:
|
|
try:
|
|
return Right[Exception, T](int(raw)) # pyright:ignore[reportArgumentType]
|
|
except Exception as e:
|
|
return Left(ValueError(f'Couldn\'t coerce `{raw}` to int: {e}'))
|
|
elif check_t is bool:
|
|
valid_trues = ['y', 'yes', 't', 'true']
|
|
valid_falses = ['n', 'no', 'f', 'false']
|
|
|
|
if raw.lower() in valid_trues:
|
|
return Right[Exception, T](True) # pyright:ignore[reportArgumentType]
|
|
elif raw.lower() in valid_falses:
|
|
return Right[Exception, T](False) # pyright:ignore[reportArgumentType]
|
|
else:
|
|
return Left(ValueError(f'Couldn\'t coerce `{raw}` to bool!'))
|
|
elif issubclass(check_t, EnumFromEnv):
|
|
return check_t.from_env(raw).lmap(
|
|
lambda exc: ValueError(f'Couldn\'t coerce `{raw}` to {check_t.__name__}: {exc}')
|
|
) # pyright:ignore[reportReturnType]
|
|
elif check_t is Option and len(args) == 1:
|
|
return _try_as_type( # pyright:ignore[reportReturnType]
|
|
raw,
|
|
args[0] # pyright:ignore[reportAny]
|
|
).map(
|
|
lambda res: Some(res)
|
|
)
|
|
else:
|
|
return Left(TypeError(f'Configuration key `{fld.name}` uses unsupported type {str(check_t)}!')) # pyright:ignore[reportUnknownArgumentType]
|
|
|
|
return (
|
|
Right[ConfigKeyError[T], str](os.environ[fld.name]) if fld.name in os.environ
|
|
else Right[ConfigKeyError[T], str](os.environ[fld.name.upper()]) if fld.name.upper() in os.environ
|
|
else Left[ConfigKeyError[T], str](ConfigKeyMissingError[T](fld))
|
|
).flat_map(
|
|
lambda vraw: _try_as_type(
|
|
vraw, fld.type # pyright:ignore[reportArgumentType]
|
|
).map(
|
|
lambda val: FieldVal[T](fld.name, val)
|
|
).lmap(
|
|
lambda exc: ConfigKeyInvalidError[T](fld, exc)
|
|
)
|
|
)
|
|
|
|
fields = [
|
|
_try_get_field(f)
|
|
for f in dataclasses.fields(cls)
|
|
]
|
|
|
|
errs, field_vals = gather_results(fields)
|
|
true_errs: list[ConfigKeyError[Any]] = [] # pyright:ignore[reportExplicitAny]
|
|
for err in errs:
|
|
# If the value has a default, it's okay if it's not in the
|
|
# environment vars
|
|
if isinstance(err, ConfigKeyMissingError):
|
|
if err.field.default_factory != dataclasses.MISSING:
|
|
log.debug(f'Populating {err.field.name} from default_factory')
|
|
field_vals.append(
|
|
FieldVal(
|
|
err.field.name,
|
|
err.field.default_factory() # pyright:ignore[reportAny]
|
|
)
|
|
)
|
|
elif err.field.default != dataclasses.MISSING:
|
|
log.debug(f'Populating {err.field.name} from default')
|
|
field_vals.append(
|
|
FieldVal(
|
|
err.field.name,
|
|
err.field.default # pyright:ignore[reportAny]
|
|
)
|
|
)
|
|
else:
|
|
true_errs.append(err)
|
|
log.error(f'Required field {err.field.name} is missing a value!')
|
|
else:
|
|
true_errs.append(err)
|
|
log.error(f'While parsing fields: {err}')
|
|
|
|
if true_errs:
|
|
return Left(true_errs)
|
|
|
|
return Right(
|
|
cls(
|
|
**{
|
|
f.key: f.val # pyright:ignore[reportAny]
|
|
for f in field_vals
|
|
}
|
|
)
|
|
)
|
|
|
|
def get_cfg() -> ConfigType:
|
|
c = ConfigType.from_env()
|
|
if isinstance(c, Left):
|
|
log.error('Unable to build config! The following errors may be the cause:')
|
|
for er in c.val:
|
|
log.error(str(er))
|
|
raise ConfigFormatError()
|
|
elif isinstance(c, Right):
|
|
return c.val
|
|
else:
|
|
# should never happen
|
|
raise TypeError()
|
|
|
|
config = get_cfg() |