Db operations: bootstrap separated to its own module.
This commit is contained in:
Leo Vasanko
@@ -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,
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user