mirror of
https://github.com/Donkie/Spoolman.git
synced 2026-10-22 18:12:18 -05:00
Donkie
ed58a6e3c4
Phase 1 could only create the owner. Every other account had to be made with python -m spoolman.cli, which meant shell access on the server was a prerequisite for having a second user at all. require_admin() has existed since phase 1 with no callers; these are its first. Two accounts are shielded from demotion, disabling and deletion, and both protections exist to stop an instance being administered into a state nobody can recover from through the interface: The owner is untouchable by anyone, themselves included. It is the account that can always fix the others, so it must not be possible to remove the last thing that can grant rights back. An administrator cannot demote, disable or delete themselves. Somebody else with the rights has to do it, which guarantees a second pair of hands still exists afterwards. Neither is a security boundary -- both are recoverable with the CLI, which remains the escape hatch when no working account is left. They are there to make the obvious mistake impossible, not to withstand an attacker who is already an administrator. The protections are consulted only on the changes that actually take something away. Granting admin is unrestricted, and re-sending an unchanged account from a form is not refused just because the form included the owner's own level -- _collect_changes compares against the stored row first, so a no-op is a no-op. Passwords may be supplied or generated. A generated one is returned exactly once, in the response that created it, which mirrors what the CLI already does and avoids an administrator inventing "Password123" for somebody. Any reset ends every session that account holds: whoever knew the old password may be precisely who the reset is defending against, and their cookie would otherwise outlive the credential it was obtained with. Disabling an account does the same, so the state in the browser agrees with the state on disk. Deleting a user takes its sessions and API keys with it and detaches its audit entries. The update path goes through a field whitelist in the data layer, so a mistake in the API layer cannot reach password_hash, is_owner, or the lockout counters. Claude-Session: https://claude.ai/code/session_01J3YgMAnSapoXqS9iU8Xnmk
817 lines
31 KiB
Python
817 lines
31 KiB
Python
"""Pydantic data models for typing the FastAPI request/responses."""
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from enum import Enum
|
|
from typing import Annotated, Literal
|
|
|
|
from pydantic import BaseModel, Field, PlainSerializer
|
|
|
|
from spoolman.auth import apikey
|
|
from spoolman.auth.levels import parse_level
|
|
from spoolman.database import models
|
|
from spoolman.math import length_from_weight
|
|
from spoolman.settings import SettingDefinition, SettingType
|
|
|
|
|
|
def datetime_to_str(dt: datetime) -> str:
|
|
"""Convert a datetime object to a string."""
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=UTC)
|
|
return dt.isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
SpoolmanDateTime = Annotated[datetime, PlainSerializer(datetime_to_str)]
|
|
|
|
|
|
def _extra_fields_description(entity: str) -> str:
|
|
r"""Build the description for an entity's ``extra`` field.
|
|
|
|
Every value in the ``extra`` map is a JSON-encoded string, regardless of the field's
|
|
configured type. For example, an ``integer`` field returns ``"42"`` (not ``42``) and a
|
|
``text`` field returns ``"\"hello\""``. Consumers must JSON-decode each value to get the
|
|
typed value. This keeps the map uniformly typed as ``dict[str, str]`` on the wire.
|
|
"""
|
|
return (
|
|
f"Extra fields for this {entity}. Every value is a JSON-encoded string, regardless of the field's "
|
|
'configured type: e.g. an integer field returns "42" (not 42) and a text field returns "\\"hello\\"". '
|
|
"Consumers must JSON-decode each value. Query the /fields endpoint for the type of each field."
|
|
)
|
|
|
|
|
|
class Message(BaseModel):
|
|
message: str = Field()
|
|
|
|
|
|
class SettingResponse(BaseModel):
|
|
value: str = Field(description="Setting value.")
|
|
is_set: bool = Field(description="Whether the setting has been set. If false, 'value' contains the default value.")
|
|
type: SettingType = Field(description="Setting type. This corresponds with JSON types.")
|
|
|
|
|
|
class SettingKV(BaseModel):
|
|
key: str = Field(description="Setting key.")
|
|
setting: SettingResponse = Field(description="Setting value.")
|
|
|
|
@staticmethod
|
|
def from_db(definition: SettingDefinition, set_value: str | None) -> SettingKV:
|
|
"""Create a new Pydantic vendor object from a database vendor object."""
|
|
return SettingKV(
|
|
key=definition.key,
|
|
setting=SettingResponse(
|
|
value=set_value if set_value is not None else definition.default,
|
|
is_set=set_value is not None,
|
|
type=definition.type,
|
|
),
|
|
)
|
|
|
|
|
|
class Vendor(BaseModel):
|
|
id: int = Field(description="Unique internal ID of this vendor.")
|
|
registered: SpoolmanDateTime = Field(description="When the vendor was registered in the database. UTC Timezone.")
|
|
name: str = Field(max_length=64, description="Vendor name.", examples=["Polymaker"])
|
|
comment: str | None = Field(
|
|
None,
|
|
max_length=1024,
|
|
description="Free text comment about this vendor.",
|
|
examples=[""],
|
|
)
|
|
empty_spool_weight: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="The empty spool weight, in grams.",
|
|
examples=[140],
|
|
)
|
|
external_id: str | None = Field(
|
|
None,
|
|
max_length=256,
|
|
description=(
|
|
"Set if this vendor comes from an external database. This contains the ID in the external database."
|
|
),
|
|
examples=["eSun"],
|
|
)
|
|
extra: dict[str, str] = Field(
|
|
description=_extra_fields_description("vendor"),
|
|
)
|
|
|
|
@staticmethod
|
|
def from_db(item: models.Vendor) -> Vendor:
|
|
"""Create a new Pydantic vendor object from a database vendor object."""
|
|
return Vendor(
|
|
id=item.id,
|
|
registered=item.registered,
|
|
name=item.name,
|
|
comment=item.comment,
|
|
empty_spool_weight=item.empty_spool_weight,
|
|
external_id=item.external_id,
|
|
extra={field.key: field.value for field in item.extra},
|
|
)
|
|
|
|
|
|
class MultiColorDirection(Enum):
|
|
"""Enum for multi-color direction."""
|
|
|
|
COAXIAL = "coaxial"
|
|
LONGITUDINAL = "longitudinal"
|
|
|
|
|
|
class Filament(BaseModel):
|
|
id: int = Field(description="Unique internal ID of this filament type.")
|
|
registered: SpoolmanDateTime = Field(description="When the filament was registered in the database. UTC Timezone.")
|
|
name: str | None = Field(
|
|
None,
|
|
max_length=64,
|
|
description=(
|
|
"Filament name, to distinguish this filament type among others from the same vendor."
|
|
"Should contain its color for example."
|
|
),
|
|
examples=["PolyTerra™ Charcoal Black"],
|
|
)
|
|
vendor: Vendor | None = Field(None, description="The vendor of this filament type.")
|
|
material: str | None = Field(
|
|
None,
|
|
max_length=64,
|
|
description="The material of this filament, e.g. PLA.",
|
|
examples=["PLA"],
|
|
)
|
|
price: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="The price of this filament in the system configured currency.",
|
|
examples=[20.0],
|
|
)
|
|
density: float = Field(gt=0, description="The density of this filament in g/cm3.", examples=[1.24])
|
|
diameter: float = Field(gt=0, description="The diameter of this filament in mm.", examples=[1.75])
|
|
weight: float | None = Field(
|
|
None,
|
|
gt=0,
|
|
description="The weight of the filament in a full spool, in grams.",
|
|
examples=[1000],
|
|
)
|
|
spool_weight: float | None = Field(None, ge=0, description="The empty spool weight, in grams.", examples=[140])
|
|
article_number: str | None = Field(
|
|
None,
|
|
max_length=64,
|
|
description="Vendor article number, e.g. EAN, QR code, etc.",
|
|
examples=["PM70820"],
|
|
)
|
|
comment: str | None = Field(
|
|
None,
|
|
max_length=1024,
|
|
description="Free text comment about this filament type.",
|
|
examples=[""],
|
|
)
|
|
settings_extruder_temp: int | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="Overridden extruder temperature, in °C.",
|
|
examples=[210],
|
|
)
|
|
settings_bed_temp: int | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="Overridden bed temperature, in °C.",
|
|
examples=[60],
|
|
)
|
|
color_hex: str | None = Field(
|
|
None,
|
|
min_length=6,
|
|
max_length=8,
|
|
description=(
|
|
"Hexadecimal color code of the filament, e.g. FF0000 for red. Supports alpha channel at the end. "
|
|
"If it's a multi-color filament, the multi_color_hexes field is used instead."
|
|
),
|
|
examples=["FF0000"],
|
|
)
|
|
multi_color_hexes: str | None = Field(
|
|
None,
|
|
min_length=6,
|
|
description=(
|
|
"Hexadecimal color code of the filament, e.g. FF0000 for red. Supports alpha channel at the end. "
|
|
"Specifying multiple colors separated by commas. "
|
|
"Also set the multi_color_direction field if you specify multiple colors."
|
|
),
|
|
examples=["FF0000,00FF00,0000FF"],
|
|
)
|
|
multi_color_direction: MultiColorDirection | None = Field(
|
|
None,
|
|
description=("Type of multi-color filament. Only set if the multi_color_hexes field is set."),
|
|
examples=["coaxial", "longitudinal"],
|
|
)
|
|
external_id: str | None = Field(
|
|
None,
|
|
max_length=256,
|
|
description=(
|
|
"Set if this filament comes from an external database. This contains the ID in the external database."
|
|
),
|
|
examples=["polymaker_pla_polysonicblack_1000_175"],
|
|
)
|
|
extra: dict[str, str] = Field(
|
|
description=_extra_fields_description("filament"),
|
|
)
|
|
|
|
@staticmethod
|
|
def from_db(item: models.Filament) -> Filament:
|
|
"""Create a new Pydantic filament object from a database filament object."""
|
|
return Filament(
|
|
id=item.id,
|
|
registered=item.registered,
|
|
name=item.name,
|
|
vendor=Vendor.from_db(item.vendor) if item.vendor is not None else None,
|
|
material=item.material,
|
|
price=item.price,
|
|
density=item.density,
|
|
diameter=item.diameter,
|
|
weight=item.weight,
|
|
spool_weight=item.spool_weight,
|
|
article_number=item.article_number,
|
|
comment=item.comment,
|
|
settings_extruder_temp=item.settings_extruder_temp,
|
|
settings_bed_temp=item.settings_bed_temp,
|
|
color_hex=item.color_hex,
|
|
multi_color_hexes=item.multi_color_hexes,
|
|
multi_color_direction=(
|
|
MultiColorDirection(item.multi_color_direction) if item.multi_color_direction is not None else None
|
|
),
|
|
external_id=item.external_id,
|
|
extra={field.key: field.value for field in item.extra},
|
|
)
|
|
|
|
|
|
class Spool(BaseModel):
|
|
id: int = Field(description="Unique internal ID of this spool of filament.")
|
|
registered: SpoolmanDateTime = Field(description="When the spool was registered in the database. UTC Timezone.")
|
|
first_used: SpoolmanDateTime | None = Field(
|
|
None,
|
|
description="First logged occurence of spool usage. UTC Timezone.",
|
|
)
|
|
last_used: SpoolmanDateTime | None = Field(
|
|
None,
|
|
description="Last logged occurence of spool usage. UTC Timezone.",
|
|
)
|
|
filament: Filament = Field(description="The filament type of this spool.")
|
|
price: float | None = Field(
|
|
None,
|
|
ge=0,
|
|
description="The price of this spool in the system configured currency.",
|
|
examples=[20.0],
|
|
)
|
|
remaining_weight: float | None = Field(
|
|
default=None,
|
|
ge=0,
|
|
description=(
|
|
"Estimated remaining weight of filament on the spool in grams. "
|
|
"Only set if the filament type has a weight set."
|
|
),
|
|
examples=[500.6],
|
|
)
|
|
initial_weight: float | None = Field(
|
|
default=None,
|
|
ge=0,
|
|
description=("The initial weight, in grams, of the filament on the spool (net weight)."),
|
|
examples=[1246],
|
|
)
|
|
spool_weight: float | None = Field(
|
|
default=None,
|
|
ge=0,
|
|
description=("Weight of an empty spool (tare weight)."),
|
|
examples=[246],
|
|
)
|
|
used_weight: float = Field(
|
|
ge=0,
|
|
description="Consumed weight of filament from the spool in grams.",
|
|
examples=[500.3],
|
|
)
|
|
remaining_length: float | None = Field(
|
|
default=None,
|
|
ge=0,
|
|
description=(
|
|
"Estimated remaining length of filament on the spool in millimeters."
|
|
" Only set if the filament type has a weight set."
|
|
),
|
|
examples=[5612.4],
|
|
)
|
|
used_length: float = Field(
|
|
ge=0,
|
|
description="Consumed length of filament from the spool in millimeters.",
|
|
examples=[50.7],
|
|
)
|
|
location: str | None = Field(
|
|
None,
|
|
max_length=64,
|
|
description="Where this spool can be found.",
|
|
examples=["Shelf A"],
|
|
)
|
|
lot_nr: str | None = Field(
|
|
None,
|
|
max_length=64,
|
|
description="Vendor manufacturing lot/batch number of the spool.",
|
|
examples=["52342"],
|
|
)
|
|
comment: str | None = Field(
|
|
None,
|
|
max_length=1024,
|
|
description="Free text comment about this specific spool.",
|
|
examples=[""],
|
|
)
|
|
archived: bool = Field(description="Whether this spool is archived and should not be used anymore.")
|
|
extra: dict[str, str] = Field(
|
|
description=_extra_fields_description("spool"),
|
|
)
|
|
|
|
@staticmethod
|
|
def from_db(item: models.Spool) -> Spool:
|
|
"""Create a new Pydantic spool object from a database spool object."""
|
|
filament = Filament.from_db(item.filament)
|
|
|
|
remaining_weight: float | None = None
|
|
remaining_length: float | None = None
|
|
|
|
if item.initial_weight is not None:
|
|
remaining_weight = max(item.initial_weight - item.used_weight, 0)
|
|
remaining_length = length_from_weight(
|
|
weight=remaining_weight,
|
|
density=filament.density,
|
|
diameter=filament.diameter,
|
|
)
|
|
elif filament.weight is not None:
|
|
remaining_weight = max(filament.weight - item.used_weight, 0)
|
|
remaining_length = length_from_weight(
|
|
weight=remaining_weight,
|
|
density=filament.density,
|
|
diameter=filament.diameter,
|
|
)
|
|
|
|
used_length = length_from_weight(
|
|
weight=item.used_weight,
|
|
density=filament.density,
|
|
diameter=filament.diameter,
|
|
)
|
|
|
|
return Spool(
|
|
id=item.id,
|
|
registered=item.registered,
|
|
first_used=item.first_used,
|
|
last_used=item.last_used,
|
|
filament=filament,
|
|
price=item.price,
|
|
initial_weight=item.initial_weight,
|
|
spool_weight=item.spool_weight,
|
|
used_weight=item.used_weight,
|
|
used_length=used_length,
|
|
remaining_weight=remaining_weight,
|
|
remaining_length=remaining_length,
|
|
location=item.location,
|
|
lot_nr=item.lot_nr,
|
|
comment=item.comment,
|
|
archived=item.archived if item.archived is not None else False,
|
|
extra={field.key: field.value for field in item.extra},
|
|
)
|
|
|
|
|
|
class SpoolGroup(BaseModel):
|
|
"""A group of spools with server-computed aggregates.
|
|
|
|
Returned by the ``/spool/group`` endpoint. Spools are grouped by one axis
|
|
(``group_by``); the aggregates are computed over the matching spools of each
|
|
group so the client can paginate whole groups without fetching every spool.
|
|
"""
|
|
|
|
group_by: str = Field(
|
|
description="The field the spools are grouped by.",
|
|
examples=["filament"],
|
|
)
|
|
key: str | None = Field(
|
|
None,
|
|
description=(
|
|
"The group key. For group_by=filament/vendor this is the entity ID as a string; for "
|
|
"material/location it is the value. Null when the grouped field is unset (e.g. spools "
|
|
"with no location or a filament with no vendor)."
|
|
),
|
|
examples=["12"],
|
|
)
|
|
spool_count: int = Field(description="Number of matching spools in this group.", examples=[6])
|
|
in_use_count: int = Field(
|
|
description="Number of matching spools that have been used (used_weight > 0).",
|
|
examples=[2],
|
|
)
|
|
total_remaining_weight: float | None = Field(
|
|
None,
|
|
description="Sum of remaining filament weight across the group's matching spools, in grams.",
|
|
examples=[3120.0],
|
|
)
|
|
last_used: SpoolmanDateTime | None = Field(
|
|
None,
|
|
description="Most recent last_used across the group's matching spools. UTC Timezone.",
|
|
)
|
|
filament: Filament | None = Field(
|
|
None,
|
|
description="The filament, embedded for group_by=filament so the header needs no extra request.",
|
|
)
|
|
vendor: Vendor | None = Field(
|
|
None,
|
|
description="The vendor, embedded for group_by=vendor.",
|
|
)
|
|
|
|
|
|
class SearchResultSpool(BaseModel):
|
|
"""A spool that matched a search, with which field matched."""
|
|
|
|
spool: Spool = Field(description="The matching spool.")
|
|
match_field: str = Field(
|
|
description=(
|
|
"Which field matched the query: a native field name (e.g. 'comment', 'location', "
|
|
"'lot_nr'), 'id' for an exact spool-id match, or 'extra.<key>' for an extra field."
|
|
),
|
|
examples=["comment"],
|
|
)
|
|
|
|
|
|
class SearchResultFilament(BaseModel):
|
|
"""A filament that matched a search, with which field matched."""
|
|
|
|
filament: Filament = Field(description="The matching filament.")
|
|
match_field: str = Field(
|
|
description=(
|
|
"Which field matched the query: a native field name (e.g. 'name', 'material', "
|
|
"'article_number', 'comment'), 'color' for a color-similarity match, or 'extra.<key>'."
|
|
),
|
|
examples=["color"],
|
|
)
|
|
|
|
|
|
class SearchResultVendor(BaseModel):
|
|
"""A vendor that matched a search, with which field matched."""
|
|
|
|
vendor: Vendor = Field(description="The matching vendor.")
|
|
match_field: str = Field(
|
|
description="Which field matched the query: 'name', 'comment', or 'extra.<key>'.",
|
|
examples=["name"],
|
|
)
|
|
|
|
|
|
class SearchResults(BaseModel):
|
|
"""Categorized results of a cross-entity search."""
|
|
|
|
spools: list[SearchResultSpool] = Field(description="Matching spools, best matches first.")
|
|
filaments: list[SearchResultFilament] = Field(description="Matching filaments, best matches first.")
|
|
vendors: list[SearchResultVendor] = Field(description="Matching vendors, best matches first.")
|
|
is_color_query: bool = Field(
|
|
description=(
|
|
"Whether the query was recognized as a color (hex code or CSS color name), in which case "
|
|
"the filament results include color-similarity matches and a threshold slider is relevant."
|
|
),
|
|
examples=[False],
|
|
)
|
|
|
|
|
|
class Info(BaseModel):
|
|
version: str = Field(examples=["0.7.0"])
|
|
debug_mode: bool = Field(examples=[False])
|
|
automatic_backups: bool = Field(examples=[True])
|
|
data_dir: str = Field(examples=["/home/app/.local/share/spoolman"])
|
|
logs_dir: str = Field(examples=["/home/app/.local/share/spoolman"])
|
|
backups_dir: str = Field(examples=["/home/app/.local/share/spoolman/backups"])
|
|
db_type: str = Field(examples=["sqlite"])
|
|
external_db_name: str = Field(
|
|
description="Display name for the external filament library, configurable via EXTERNAL_DB_NAME.",
|
|
examples=["SpoolmanDB"],
|
|
)
|
|
git_commit: str | None = Field(None, examples=["a1b2c3d"])
|
|
build_date: SpoolmanDateTime | None = Field(None, examples=["2021-01-01T00:00:00Z"])
|
|
|
|
|
|
class HealthCheck(BaseModel):
|
|
status: str = Field(examples=["healthy"])
|
|
|
|
|
|
class BackupResponse(BaseModel):
|
|
path: str = Field(
|
|
default=None,
|
|
description="Path to the created backup file.",
|
|
examples=["/home/app/.local/share/spoolman/backups/spoolman.db"],
|
|
)
|
|
|
|
|
|
class EventType(str, Enum):
|
|
"""Event types."""
|
|
|
|
ADDED = "added"
|
|
UPDATED = "updated"
|
|
DELETED = "deleted"
|
|
|
|
|
|
class Event(BaseModel):
|
|
"""Event."""
|
|
|
|
type: EventType = Field(description="Event type.")
|
|
resource: str = Field(description="Resource type.")
|
|
date: SpoolmanDateTime = Field(description="When the event occured. UTC Timezone.")
|
|
payload: BaseModel
|
|
|
|
|
|
class SpoolEvent(Event):
|
|
"""Event."""
|
|
|
|
payload: Spool = Field(description="Updated spool.")
|
|
resource: Literal["spool"] = Field(description="Resource type.")
|
|
|
|
|
|
class FilamentEvent(Event):
|
|
"""Event."""
|
|
|
|
payload: Filament = Field(description="Updated filament.")
|
|
resource: Literal["filament"] = Field(description="Resource type.")
|
|
|
|
|
|
class VendorEvent(Event):
|
|
"""Event."""
|
|
|
|
payload: Vendor = Field(description="Updated vendor.")
|
|
resource: Literal["vendor"] = Field(description="Resource type.")
|
|
|
|
|
|
class SettingEvent(Event):
|
|
"""Event."""
|
|
|
|
payload: SettingKV = Field(description="Updated setting.")
|
|
resource: Literal["setting"] = Field(description="Resource type.")
|
|
|
|
|
|
#
|
|
# Authentication.
|
|
#
|
|
# These describe the /auth endpoints. They are additive: nothing above changes shape,
|
|
# and none of this is reachable in a meaningful way unless SPOOLMAN_AUTH_ENABLED is set.
|
|
#
|
|
|
|
# Short enough to stay usable for a homelab instance, long enough that the lockout in
|
|
# spoolman/database/auth_user.py is the binding constraint rather than the search space.
|
|
MIN_PASSWORD_LENGTH = 10
|
|
MAX_PASSWORD_LENGTH = 1024
|
|
MAX_USERNAME_LENGTH = 64
|
|
MAX_DISPLAY_NAME_LENGTH = 128
|
|
|
|
|
|
class AuthOidcConfig(BaseModel):
|
|
"""OIDC provider advertisement. Phase 3; always absent for now."""
|
|
|
|
enabled: bool = Field(default=False, description="Whether an OIDC provider is configured.")
|
|
name: str | None = Field(default=None, description="Display name for the sign-in button.")
|
|
|
|
|
|
class AuthConfig(BaseModel):
|
|
"""What a client needs to know before anyone has signed in."""
|
|
|
|
enabled: bool = Field(description="Whether authentication is enforced on this instance.")
|
|
setup_required: bool = Field(
|
|
description="Whether this instance is unclaimed. When true, anyone may claim ownership.",
|
|
)
|
|
anonymous_read: bool = Field(description="Whether unauthenticated clients may read.")
|
|
oidc: AuthOidcConfig = Field(default_factory=AuthOidcConfig, description="OIDC provider info.")
|
|
mtls: bool = Field(default=False, description="Whether client certificates are accepted. Phase 5.")
|
|
|
|
|
|
class AuthUserInfo(BaseModel):
|
|
"""A user account, as shown to itself or an administrator."""
|
|
|
|
id: int = Field(description="Unique internal ID of this user.")
|
|
username: str = Field(description="Username, stored lowercased.")
|
|
display_name: str | None = Field(default=None, description="Friendly name, if set.")
|
|
level: str = Field(description="Permission level: read, edit or manage.", examples=["edit"])
|
|
is_admin: bool = Field(description="Whether this user administers other users.")
|
|
is_owner: bool = Field(description="Whether this user owns the instance.")
|
|
is_active: bool = Field(default=True, description="Whether the account may sign in at all.")
|
|
must_change_password: bool = Field(description="Whether a password change is required.")
|
|
|
|
@staticmethod
|
|
def from_db(user: models.AuthUser) -> AuthUserInfo:
|
|
"""Create a Pydantic user object from a database user object."""
|
|
return AuthUserInfo(
|
|
id=user.id,
|
|
username=user.username,
|
|
display_name=user.display_name,
|
|
level=user.level,
|
|
is_admin=user.is_admin,
|
|
is_owner=user.is_owner,
|
|
is_active=user.is_active,
|
|
must_change_password=user.must_change_password,
|
|
)
|
|
|
|
|
|
class AuthSessionInfo(BaseModel):
|
|
"""The caller's current standing.
|
|
|
|
Always answers rather than rejecting, so a signed-out client can tell "no session"
|
|
apart from "server unreachable" without treating a 401 as an error.
|
|
"""
|
|
|
|
authenticated: bool = Field(description="Whether a user is signed in.")
|
|
anonymous: bool = Field(description="Whether access comes from the anonymous-read setting.")
|
|
level: str = Field(description="Effective permission level. 'manage' when auth is disabled.")
|
|
is_admin: bool = Field(description="Whether the caller administers other users.")
|
|
is_owner: bool = Field(description="Whether the caller owns the instance.")
|
|
user: AuthUserInfo | None = Field(default=None, description="The signed-in user, if any.")
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
"""Credentials for a sign-in attempt."""
|
|
|
|
username: str = Field(min_length=1, max_length=MAX_USERNAME_LENGTH, description="Username.")
|
|
password: str = Field(min_length=1, max_length=MAX_PASSWORD_LENGTH, description="Password.")
|
|
remember: bool = Field(default=False, description="Whether the session should outlive the browser session.")
|
|
|
|
|
|
class SetupRequest(BaseModel):
|
|
"""The owner account to create when claiming an unclaimed instance."""
|
|
|
|
username: str = Field(min_length=1, max_length=MAX_USERNAME_LENGTH, description="Username for the owner.")
|
|
password: str = Field(
|
|
min_length=MIN_PASSWORD_LENGTH,
|
|
max_length=MAX_PASSWORD_LENGTH,
|
|
description="Password for the owner.",
|
|
)
|
|
display_name: str | None = Field(default=None, max_length=MAX_DISPLAY_NAME_LENGTH, description="Friendly name.")
|
|
|
|
|
|
class PasswordChangeRequest(BaseModel):
|
|
"""A request to replace one's own password."""
|
|
|
|
current_password: str = Field(min_length=1, max_length=MAX_PASSWORD_LENGTH, description="The current password.")
|
|
new_password: str = Field(
|
|
min_length=MIN_PASSWORD_LENGTH,
|
|
max_length=MAX_PASSWORD_LENGTH,
|
|
description="The new password.",
|
|
)
|
|
|
|
|
|
#
|
|
# Authentication, phase 2: API keys, user administration and the audit log.
|
|
#
|
|
|
|
MAX_KEY_NAME_LENGTH = 64
|
|
MAX_AUDIT_EVENT_LENGTH = 64
|
|
|
|
# The longest an API key may be issued for. Mirrors
|
|
# spoolman.database.auth_api_key.MAX_EXPIRY_DAYS; validated here so a bad value is a 422
|
|
# from the schema rather than a database error.
|
|
MAX_KEY_EXPIRY_DAYS = 3650
|
|
|
|
|
|
class AuthApiKeyInfo(BaseModel):
|
|
"""An API key, as shown to its owner.
|
|
|
|
Never carries the secret. The complete key exists in a readable form exactly once,
|
|
in the response to the request that created it.
|
|
"""
|
|
|
|
id: int = Field(description="Unique internal ID of this key.")
|
|
name: str = Field(description="The label its owner gave it.")
|
|
level: str = Field(description="Permission level the key was issued with.", examples=["edit"])
|
|
effective_level: str = Field(
|
|
description="What the key can actually do right now, after capping against its owner's level.",
|
|
examples=["read"],
|
|
)
|
|
prefix: str = Field(description="The public half of the key, enough to recognise it in a log.")
|
|
created: SpoolmanDateTime = Field(description="When the key was issued.")
|
|
expires: SpoolmanDateTime | None = Field(default=None, description="When the key stops working, if ever.")
|
|
last_used: SpoolmanDateTime | None = Field(
|
|
default=None,
|
|
description="When the key was last accepted. Updated at most once every few minutes.",
|
|
)
|
|
revoked: bool = Field(description="Whether the key has been withdrawn.")
|
|
expired: bool = Field(description="Whether the key's expiry has passed.")
|
|
|
|
@staticmethod
|
|
def from_db(key: models.AuthApiKey, owner_level: str) -> AuthApiKeyInfo:
|
|
"""Create a Pydantic API key object from a database API key object."""
|
|
return AuthApiKeyInfo(
|
|
id=key.id,
|
|
name=key.name,
|
|
level=key.level,
|
|
effective_level=str(apikey.effective_level(parse_level(key.level), parse_level(owner_level))),
|
|
prefix=key.prefix,
|
|
created=key.created,
|
|
expires=key.expires,
|
|
last_used=key.last_used,
|
|
revoked=key.revoked,
|
|
expired=key.expires is not None and key.expires <= datetime.utcnow(),
|
|
)
|
|
|
|
|
|
class AuthApiKeyCreated(BaseModel):
|
|
"""A newly issued API key, including the secret."""
|
|
|
|
key: str = Field(description="The complete key. Shown once and never again; store it now.")
|
|
info: AuthApiKeyInfo = Field(description="The key's metadata, as it will appear in listings.")
|
|
|
|
|
|
class ApiKeyCreateRequest(BaseModel):
|
|
"""A request to issue an API key."""
|
|
|
|
name: str = Field(min_length=1, max_length=MAX_KEY_NAME_LENGTH, description="A label, to tell keys apart.")
|
|
level: str = Field(
|
|
description="Permission level to issue at. May not exceed the creator's own level.",
|
|
examples=["edit"],
|
|
)
|
|
expires_days: int | None = Field(
|
|
default=None,
|
|
ge=1,
|
|
le=MAX_KEY_EXPIRY_DAYS,
|
|
description="How many days the key should live for. Omit for a key that never expires.",
|
|
)
|
|
|
|
|
|
class AuthUserCreateRequest(BaseModel):
|
|
"""An account for an administrator to create."""
|
|
|
|
username: str = Field(min_length=1, max_length=MAX_USERNAME_LENGTH, description="Username, stored lowercased.")
|
|
password: str | None = Field(
|
|
default=None,
|
|
min_length=MIN_PASSWORD_LENGTH,
|
|
max_length=MAX_PASSWORD_LENGTH,
|
|
description="Initial password. Omit to have one generated and returned once.",
|
|
)
|
|
display_name: str | None = Field(default=None, max_length=MAX_DISPLAY_NAME_LENGTH, description="Friendly name.")
|
|
level: str = Field(default="read", description="Permission level to grant.", examples=["edit"])
|
|
is_admin: bool = Field(default=False, description="Whether the account administers other users.")
|
|
must_change_password: bool = Field(
|
|
default=True,
|
|
description="Whether the user must choose a new password at next sign-in.",
|
|
)
|
|
|
|
|
|
class AuthUserUpdateRequest(BaseModel):
|
|
"""Changes to an existing account. Omitted fields are left alone."""
|
|
|
|
display_name: str | None = Field(default=None, max_length=MAX_DISPLAY_NAME_LENGTH, description="Friendly name.")
|
|
level: str | None = Field(default=None, description="Permission level to grant.", examples=["manage"])
|
|
is_admin: bool | None = Field(default=None, description="Whether the account administers other users.")
|
|
is_active: bool | None = Field(default=None, description="Whether the account may sign in at all.")
|
|
|
|
|
|
class AuthPasswordResetRequest(BaseModel):
|
|
"""An administrator's replacement of somebody else's password."""
|
|
|
|
password: str | None = Field(
|
|
default=None,
|
|
min_length=MIN_PASSWORD_LENGTH,
|
|
max_length=MAX_PASSWORD_LENGTH,
|
|
description="The new password. Omit to have one generated and returned once.",
|
|
)
|
|
must_change_password: bool = Field(
|
|
default=True,
|
|
description="Whether the user must choose a new password at next sign-in.",
|
|
)
|
|
|
|
|
|
class AuthUserCreated(BaseModel):
|
|
"""A newly created account, with its password if one was generated."""
|
|
|
|
user: AuthUserInfo = Field(description="The account.")
|
|
password: str | None = Field(
|
|
default=None,
|
|
description="The generated password, present only when the caller did not supply one. Shown once.",
|
|
)
|
|
|
|
|
|
class AuthAuditEntry(BaseModel):
|
|
"""One line of the audit trail."""
|
|
|
|
id: int = Field(description="Unique internal ID of this entry.")
|
|
date: SpoolmanDateTime = Field(description="When it happened, in UTC.")
|
|
event: str = Field(description="What happened.", examples=["login.success"])
|
|
actor_user_id: int | None = Field(default=None, description="Which account acted, if it still exists.")
|
|
actor_username: str | None = Field(default=None, description="That account's username, resolved for display.")
|
|
actor_kind: str = Field(description="How the actor was authenticated.", examples=["user"])
|
|
target: str | None = Field(default=None, description="What was acted on.")
|
|
ip: str | None = Field(default=None, description="Where the request came from.")
|
|
user_agent: str | None = Field(default=None, description="What the request identified itself as.")
|
|
detail: dict | None = Field(default=None, description="Anything else worth keeping.")
|
|
|
|
@staticmethod
|
|
def from_db(entry: models.AuthAuditLog, actor_username: str | None = None) -> AuthAuditEntry:
|
|
"""Create a Pydantic audit entry from a database audit entry."""
|
|
detail = None
|
|
if entry.detail:
|
|
try:
|
|
parsed = json.loads(entry.detail)
|
|
except ValueError:
|
|
# A row written by a future version, or corrupted. Showing the raw text
|
|
# beats dropping the entry: the point of an audit log is that entries do
|
|
# not silently vanish.
|
|
parsed = {"raw": entry.detail}
|
|
detail = parsed if isinstance(parsed, dict) else {"value": parsed}
|
|
return AuthAuditEntry(
|
|
id=entry.id,
|
|
date=entry.date,
|
|
event=entry.event,
|
|
actor_user_id=entry.actor_user_id,
|
|
actor_username=actor_username,
|
|
actor_kind=entry.actor_kind,
|
|
target=entry.target,
|
|
ip=entry.ip,
|
|
user_agent=entry.user_agent,
|
|
detail=detail,
|
|
)
|