Db operations: bootstrap separated to its own module.

This commit is contained in:
2026-02-10 22:08:54 +00:00
parent c2ea01e6d9
commit 2237e6b5e9
4 changed files with 125 additions and 119 deletions
+1 -1
View File
@@ -26,10 +26,10 @@ from paskia.db.background import (
stop_background, stop_background,
stop_cleanup, stop_cleanup,
) )
from paskia.db.bootstrap import bootstrap
from paskia.db.operations import ( from paskia.db.operations import (
add_permission_to_org, add_permission_to_org,
add_permission_to_role, add_permission_to_role,
bootstrap,
cleanup_expired, cleanup_expired,
create_credential, create_credential,
create_credential_session, create_credential_session,
+122
View File
@@ -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
-117
View File
@@ -709,123 +709,6 @@ def create_credential_session(
return session.key 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 # Config operations
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
+2 -1
View File
@@ -28,6 +28,7 @@ from paskia.db import (
Permission, Permission,
Role, Role,
User, User,
bootstrap,
create_credential, create_credential,
create_reset_token, create_reset_token,
create_role, create_role,
@@ -67,7 +68,7 @@ async def test_db() -> AsyncGenerator[DB, None]:
ops_db._db = db ops_db._db = db
ops_db._store = store ops_db._store = store
# Bootstrap creates the initial permissions, org, role, and admin user # Bootstrap creates the initial permissions, org, role, and admin user
ops_db.bootstrap( bootstrap(
org_name="Test Organization", org_name="Test Organization",
admin_name="Test Admin", admin_name="Test Admin",
) )