From f0d360758cc208cfebd7d7210830fc3d7959fd60 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Tue, 10 Feb 2026 22:08:54 +0000 Subject: [PATCH] Db operations: bootstrap separated to its own module. --- paskia/db/__init__.py | 2 +- paskia/db/bootstrap.py | 122 ++++++++++++++++++++++++++++++++++++++++ paskia/db/operations.py | 117 -------------------------------------- tests/conftest.py | 3 +- 4 files changed, 125 insertions(+), 119 deletions(-) create mode 100644 paskia/db/bootstrap.py diff --git a/paskia/db/__init__.py b/paskia/db/__init__.py index 973860d..1f7ac5c 100644 --- a/paskia/db/__init__.py +++ b/paskia/db/__init__.py @@ -26,10 +26,10 @@ from paskia.db.background import ( stop_background, stop_cleanup, ) +from paskia.db.bootstrap import bootstrap from paskia.db.operations import ( add_permission_to_org, add_permission_to_role, - bootstrap, cleanup_expired, create_credential, create_credential_session, diff --git a/paskia/db/bootstrap.py b/paskia/db/bootstrap.py new file mode 100644 index 0000000..89ffe63 --- /dev/null +++ b/paskia/db/bootstrap.py @@ -0,0 +1,122 @@ +""" +Bootstrap operations for initial system setup. +""" + +from datetime import UTC, datetime + +import uuid7 + +import paskia.db.operations as _ops +from paskia.db.structs import Config, Org, Permission, ResetToken, Role, User + + +def bootstrap( + org_name: str = "Organization", + admin_name: str = "Admin", + reset_passphrase: str | None = None, + reset_expiry: datetime | None = None, + config: Config | None = None, +) -> str: + """Bootstrap the entire system in a single transaction. + + Creates: + - auth:admin permission (Master Admin) + - auth:org:admin permission (Org Admin) + - Organization with Administration role + - Admin user with Administration role + - Reset token for admin registration + - Config (if provided) + + This is the only way to create a new database file. + All data is created atomically - if any step fails, nothing is written. + + Args: + org_name: Display name for the organization (default: "Organization") + admin_name: Display name for the admin user (default: "Admin") + reset_passphrase: Passphrase for the reset token (generated if not provided) + reset_expiry: Expiry datetime for the reset token (default: 14 days) + config: Configuration to store (rp_id, rp_name, origins, etc.) + + Returns: + The reset passphrase for admin registration. + """ + + # Check if system is already bootstrapped + for p in _ops._db.permissions.values(): + if p.scope == "auth:admin": + raise ValueError( + "System already bootstrapped (auth:admin permission exists)" + ) + + # Generate UUIDs upfront + now = datetime.now(UTC) + perm_admin_uuid = uuid7.create(now) + perm_org_admin_uuid = uuid7.create(now) + org_uuid = uuid7.create(now) + role_uuid = uuid7.create(now) + user_uuid = uuid7.create(now) + + # Set reset token expiry (passphrase generated by ResetToken.create) + if reset_expiry is None: + from paskia.authsession import reset_expires # noqa: PLC0415 + + reset_expiry = reset_expires() + + with _ops._db.transaction("bootstrap"): + # Create auth:admin permission + perm_admin = Permission( + scope="auth:admin", + display_name="Master Admin", + orgs={org_uuid: True}, # Grant to org + ) + perm_admin.uuid = perm_admin_uuid + _ops._db.permissions[perm_admin_uuid] = perm_admin + + # Create auth:org:admin permission + perm_org_admin = Permission( + scope="auth:org:admin", + display_name="Org Admin", + orgs={org_uuid: True}, # Grant to org + ) + perm_org_admin.uuid = perm_org_admin_uuid + _ops._db.permissions[perm_org_admin_uuid] = perm_org_admin + + # Create organization + new_org = Org.create(display_name=org_name) + new_org.uuid = org_uuid + _ops._db.orgs[org_uuid] = new_org + + # Create Administration role with both permissions + admin_role = Role( + org_uuid=org_uuid, + display_name="Administration", + permissions={perm_admin_uuid: True, perm_org_admin_uuid: True}, + ) + admin_role.uuid = role_uuid + _ops._db.roles[role_uuid] = admin_role + + # Create admin user + admin_user = User( + display_name=admin_name, + role_uuid=role_uuid, + created_at=now, + last_seen=None, + visits=0, + ) + admin_user.uuid = user_uuid + _ops._db.users[user_uuid] = admin_user + + # Create reset token + reset_token, reset_passphrase = ResetToken.create( + user=user_uuid, + expiry=reset_expiry, + token_type="admin bootstrap", + passphrase=reset_passphrase, + ) + _ops._db.reset_tokens[reset_token.key] = reset_token + + # Set config if provided + if config is not None: + _ops._db.config = config + + return reset_passphrase diff --git a/paskia/db/operations.py b/paskia/db/operations.py index c22239f..a4e303d 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -709,123 +709,6 @@ def create_credential_session( return session.key -# ------------------------------------------------------------------------- -# Bootstrap (single transaction for initial system setup) -# ------------------------------------------------------------------------- - - -def bootstrap( - org_name: str = "Organization", - admin_name: str = "Admin", - reset_passphrase: str | None = None, - reset_expiry: datetime | None = None, - config: Config | None = None, -) -> str: - """Bootstrap the entire system in a single transaction. - - Creates: - - auth:admin permission (Master Admin) - - auth:org:admin permission (Org Admin) - - Organization with Administration role - - Admin user with Administration role - - Reset token for admin registration - - Config (if provided) - - This is the only way to create a new database file. - All data is created atomically - if any step fails, nothing is written. - - Args: - org_name: Display name for the organization (default: "Organization") - admin_name: Display name for the admin user (default: "Admin") - reset_passphrase: Passphrase for the reset token (generated if not provided) - reset_expiry: Expiry datetime for the reset token (default: 14 days) - config: Configuration to store (rp_id, rp_name, origins, etc.) - - Returns: - The reset passphrase for admin registration. - """ - - # Check if system is already bootstrapped - for p in _db.permissions.values(): - if p.scope == "auth:admin": - raise ValueError( - "System already bootstrapped (auth:admin permission exists)" - ) - - # Generate UUIDs upfront - now = datetime.now(UTC) - perm_admin_uuid = uuid7.create(now) - perm_org_admin_uuid = uuid7.create(now) - org_uuid = uuid7.create(now) - role_uuid = uuid7.create(now) - user_uuid = uuid7.create(now) - - # Set reset token expiry (passphrase generated by ResetToken.create) - if reset_expiry is None: - from paskia.authsession import reset_expires # noqa: PLC0415 - - reset_expiry = reset_expires() - - with _db.transaction("bootstrap"): - # Create auth:admin permission - perm_admin = Permission( - scope="auth:admin", - display_name="Master Admin", - orgs={org_uuid: True}, # Grant to org - ) - perm_admin.uuid = perm_admin_uuid - _db.permissions[perm_admin_uuid] = perm_admin - - # Create auth:org:admin permission - perm_org_admin = Permission( - scope="auth:org:admin", - display_name="Org Admin", - orgs={org_uuid: True}, # Grant to org - ) - perm_org_admin.uuid = perm_org_admin_uuid - _db.permissions[perm_org_admin_uuid] = perm_org_admin - - # Create organization - new_org = Org.create(display_name=org_name) - new_org.uuid = org_uuid - _db.orgs[org_uuid] = new_org - - # Create Administration role with both permissions - admin_role = Role( - org_uuid=org_uuid, - display_name="Administration", - permissions={perm_admin_uuid: True, perm_org_admin_uuid: True}, - ) - admin_role.uuid = role_uuid - _db.roles[role_uuid] = admin_role - - # Create admin user - admin_user = User( - display_name=admin_name, - role_uuid=role_uuid, - created_at=now, - last_seen=None, - visits=0, - ) - admin_user.uuid = user_uuid - _db.users[user_uuid] = admin_user - - # Create reset token - reset_token, reset_passphrase = ResetToken.create( - user=user_uuid, - expiry=reset_expiry, - token_type="admin bootstrap", - passphrase=reset_passphrase, - ) - _db.reset_tokens[reset_token.key] = reset_token - - # Set config if provided - if config is not None: - _db.config = config - - return reset_passphrase - - # ------------------------------------------------------------------------- # Config operations # ------------------------------------------------------------------------- diff --git a/tests/conftest.py b/tests/conftest.py index 1900147..c0493b4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -28,6 +28,7 @@ from paskia.db import ( Permission, Role, User, + bootstrap, create_credential, create_reset_token, create_role, @@ -67,7 +68,7 @@ async def test_db() -> AsyncGenerator[DB, None]: ops_db._db = db ops_db._store = store # Bootstrap creates the initial permissions, org, role, and admin user - ops_db.bootstrap( + bootstrap( org_name="Test Organization", admin_name="Test Admin", )