diff --git a/paskia/authsession.py b/paskia/authsession.py index c322d61..f9224a2 100644 --- a/paskia/authsession.py +++ b/paskia/authsession.py @@ -9,13 +9,16 @@ independent of any web framework: """ from datetime import datetime, timezone +from typing import TYPE_CHECKING from uuid import UUID from paskia import db -from paskia.config import SESSION_LIFETIME -from paskia.db import ResetToken, Session +from paskia.config import RESET_LIFETIME, SESSION_LIFETIME from paskia.util import hostutil +if TYPE_CHECKING: + from paskia.db import ResetToken, Session + EXPIRES = SESSION_LIFETIME @@ -24,21 +27,21 @@ def expires() -> datetime: def reset_expires() -> datetime: - from .config import RESET_LIFETIME - return datetime.now(timezone.utc) + RESET_LIFETIME -async def get_reset(token: str) -> ResetToken: +async def get_reset(token: str) -> "ResetToken": """Validate a credential reset token.""" + record = db.get_reset_token(token) if record: return record raise ValueError("This authentication link is no longer valid.") -async def get_session(token: str, host: str | None = None) -> Session: +async def get_session(token: str, host: str | None = None) -> "Session": """Validate a session token and return session data if valid.""" + host = hostutil.normalize_host(host) if not host: raise ValueError("Invalid host") diff --git a/paskia/bootstrap.py b/paskia/bootstrap.py index 41c3cb2..4ba0123 100644 --- a/paskia/bootstrap.py +++ b/paskia/bootstrap.py @@ -9,8 +9,8 @@ generating a reset link for initial admin setup. import asyncio import logging -from paskia import db -from paskia.util import hostutil +from paskia import authsession, db, globals +from paskia.util import hostutil, passphrase logger = logging.getLogger(__name__) @@ -72,8 +72,6 @@ async def check_admin_credentials() -> bool: if not db.get_user_credential_ids(admin_user.uuid): # Admin exists but has no credentials, create reset link - from paskia import authsession - from paskia.util import passphrase token = passphrase.generate() expiry = authsession.reset_expires() @@ -118,8 +116,6 @@ async def main(): # Configure logging for CLI usage logging.basicConfig(level=logging.INFO, format="%(message)s", force=True) - from paskia import globals - await globals.init() diff --git a/paskia/db/__init__.py b/paskia/db/__init__.py index ddebe67..a69a9d5 100644 --- a/paskia/db/__init__.py +++ b/paskia/db/__init__.py @@ -19,6 +19,7 @@ Usage: db.create_user(user) """ +import paskia.db.operations as operations from paskia.db.background import ( start_background, start_cleanup, @@ -81,9 +82,7 @@ from paskia.db.structs import ( def data() -> DB: """Get the database instance for direct read access.""" - from paskia.db.operations import _db - - return _db + return operations._db __all__ = [ diff --git a/paskia/db/background.py b/paskia/db/background.py index 60b568e..8d7f950 100644 --- a/paskia/db/background.py +++ b/paskia/db/background.py @@ -8,6 +8,8 @@ import asyncio import logging from datetime import datetime, timezone +from paskia.db.operations import _db, _store + # Flush changes to disk every N seconds FLUSH_INTERVAL = 1 # Cleanup expired items every N seconds (cheap when nothing to remove) @@ -20,7 +22,6 @@ _background_task: asyncio.Task | None = None def cleanup() -> None: """Remove expired sessions and reset tokens from the database.""" - from paskia.db.operations import _db if _db is None: return @@ -45,7 +46,6 @@ def cleanup() -> None: async def flush() -> None: """Write all pending database changes to disk.""" - from paskia.db.operations import _store if _store is None: _logger.warning("flush() called but _store is None") diff --git a/paskia/db/operations.py b/paskia/db/operations.py index c194911..8a4458b 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -13,6 +13,9 @@ import secrets from datetime import datetime, timezone from uuid import UUID +import uuid7 + +from paskia.config import SESSION_LIFETIME from paskia.db.jsonl import ( DB_PATH_DEFAULT, JsonlStore, @@ -28,6 +31,8 @@ from paskia.db.structs import ( SessionContext, User, ) +from paskia.util.hostutil import normalize_host +from paskia.util.passphrase import generate as generate_passphrase from paskia.util.passphrase import is_well_formed as _is_passphrase _logger = logging.getLogger(__name__) @@ -137,7 +142,6 @@ def get_session_context( - Example usage in docstring (db/__init__.py:16) - Get session context from auth token (util/permutil.py:43) """ - from paskia.util.hostutil import normalize_host if session_key not in _db.sessions: return None @@ -279,7 +283,6 @@ def create_organization(org: Org, *, ctx: SessionContext | None = None) -> None: _db.orgs[org.uuid] = new_org new_org.uuid = org.uuid # Create Administration role with org admin permission - import uuid7 admin_role_uuid = uuid7.create() # Find the auth:org:admin permission UUID @@ -786,7 +789,6 @@ def create_credential_session( Returns the generated session token. """ - from paskia.config import SESSION_LIFETIME now = datetime.now(timezone.utc) expiry = now + SESSION_LIFETIME @@ -853,10 +855,6 @@ def bootstrap( Returns: The reset passphrase for admin registration. """ - import uuid7 - - from paskia.authsession import reset_expires - from paskia.util.passphrase import generate as generate_passphrase # Check if system is already bootstrapped for p in _db.permissions.values(): @@ -876,6 +874,7 @@ def bootstrap( if reset_passphrase is None: reset_passphrase = generate_passphrase() if reset_expiry is None: + from paskia.util.timeutil import reset_expires # noqa: PLC0415 reset_expiry = reset_expires() reset_key = _reset_key(reset_passphrase) diff --git a/paskia/fastapi/admin.py b/paskia/fastapi/admin.py index e7f0a5d..e039521 100644 --- a/paskia/fastapi/admin.py +++ b/paskia/fastapi/admin.py @@ -5,10 +5,16 @@ from uuid import UUID from fastapi import Body, FastAPI, HTTPException, Query, Request, Response from fastapi.responses import JSONResponse +from paskia import aaguid as aaguid_mod from paskia import db from paskia.authsession import EXPIRES, reset_expires +from paskia.db import Org as OrgDC +from paskia.db import Permission as PermDC +from paskia.db import Role as RoleDC +from paskia.db import User as UserDC from paskia.fastapi import authz from paskia.fastapi.session import AUTH_COOKIE +from paskia.globals import passkey from paskia.util import ( hostutil, passphrase, @@ -17,6 +23,7 @@ from paskia.util import ( useragent, vitedev, ) +from paskia.util.hostutil import normalize_host app = FastAPI() @@ -137,7 +144,6 @@ async def admin_create_org( ctx = await authz.verify( auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all ) - from ..db import Org as OrgDC # local import to avoid cycles display_name = payload.get("display_name") or "New Organization" permissions = payload.get("permissions") or [] @@ -296,7 +302,6 @@ async def admin_create_role( raise authz.AuthException( status_code=403, detail="Insufficient permissions", mode="forbidden" ) - from ..db import Role as RoleDC display_name = payload.get("display_name") or "New Role" perms = payload.get("permissions") or [] @@ -486,7 +491,6 @@ async def admin_create_user( role_name = payload.get("role") if not display_name or not role_name: raise ValueError("display_name and role are required") - from ..db import User as UserDC roles = [r for r in db.data().roles.values() if r.org == org_uuid] role_obj = next((r for r in roles if r.display_name == role_name), None) @@ -674,7 +678,6 @@ async def admin_get_user_detail( "sign_count": c.sign_count, } ) - from .. import aaguid as aaguid_mod aaguid_info = aaguid_mod.filter(aaguids) @@ -854,7 +857,6 @@ def _validate_permission_domain(domain: str | None) -> None: """Validate that domain is rp_id or a subdomain of it.""" if domain is None: return - from paskia.globals import passkey rp_id = passkey.instance.rp_id if domain == rp_id or domain.endswith(f".{rp_id}"): @@ -870,7 +872,6 @@ def _check_admin_lockout( Raises ValueError if this change would result in no auth:admin permissions being accessible from the current host. """ - from paskia.util.hostutil import normalize_host normalized_host = normalize_host(current_host) host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None @@ -905,7 +906,6 @@ def _check_admin_lockout_on_delete(perm_uuid: str, current_host: str | None) -> Raises ValueError if this deletion would result in no auth:admin permissions being accessible from the current host. """ - from paskia.util.hostutil import normalize_host normalized_host = normalize_host(current_host) host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None @@ -970,7 +970,6 @@ async def admin_create_permission( match=permutil.has_all, max_age="5m", ) - from ..db import Permission as PermDC scope = payload.get("scope") or payload.get( "id" @@ -1018,8 +1017,6 @@ async def admin_update_permission( if perm.scope == "auth:admin" or new_scope == "auth:admin": _check_admin_lockout(str(perm.uuid), domain_value, request.headers.get("host")) - from ..db import Permission as PermDC - updated_perm = PermDC( scope=new_scope, display_name=new_display_name, diff --git a/paskia/fastapi/mainapp.py b/paskia/fastapi/mainapp.py index 3e134b7..22337d8 100644 --- a/paskia/fastapi/mainapp.py +++ b/paskia/fastapi/mainapp.py @@ -1,3 +1,4 @@ +import json import logging import os from contextlib import asynccontextmanager @@ -7,6 +8,7 @@ from fastapi import FastAPI, HTTPException, Request, Response from fastapi.responses import FileResponse, RedirectResponse from fastapi_vue import Frontend +from paskia import globals from paskia.fastapi import admin, api, auth_host, ws from paskia.fastapi.session import AUTH_COOKIE from paskia.util import hostutil, passphrase, vitedev @@ -30,9 +32,6 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path so that uvicorn reload / multiprocess workers inherit the settings. All keys are guaranteed to exist; values are already normalized by __main__.py. """ - import json - - from paskia import globals config = json.loads(os.environ["PASKIA_CONFIG"]) diff --git a/paskia/fastapi/remote.py b/paskia/fastapi/remote.py index f387fbc..8cf0494 100644 --- a/paskia/fastapi/remote.py +++ b/paskia/fastapi/remote.py @@ -16,10 +16,11 @@ import base64url from fastapi import FastAPI, WebSocket, WebSocketDisconnect from paskia import db, remoteauth +from paskia.authsession import expires from paskia.fastapi.session import infodict from paskia.fastapi.wschat import authenticate_chat from paskia.fastapi.wsutil import validate_origin, websocket_error_handler -from paskia.util import passphrase, pow +from paskia.util import hostutil, passphrase, pow, useragent # Create a FastAPI subapp for remote auth WebSocket endpoints app = FastAPI() @@ -268,7 +269,6 @@ async def websocket_remote_auth_permit(ws: WebSocket): 6. Client sends WebAuthn response 7. Server sends {status: "success", message: "..."} """ - from paskia.util import useragent origin = validate_origin(ws) @@ -289,7 +289,6 @@ async def websocket_remote_auth_permit(ws: WebSocket): ) request = None - webauthn_challenge = None explicitly_denied = False try: @@ -321,8 +320,6 @@ async def websocket_remote_auth_permit(ws: WebSocket): if request.action == "register": # For registration, create a reset token for device addition - from paskia.authsession import expires - from paskia.util import hostutil token_str = passphrase.generate() expiry = expires() @@ -345,8 +342,6 @@ async def websocket_remote_auth_permit(ws: WebSocket): ) else: # Default login action - from paskia.authsession import expires - from paskia.util import hostutil normalized_host = hostutil.normalize_host(request.host) session_token = db.login( diff --git a/paskia/globals.py b/paskia/globals.py index 2cfe983..5c0d7d5 100644 --- a/paskia/globals.py +++ b/paskia/globals.py @@ -1,5 +1,8 @@ from typing import Generic, TypeVar +from paskia import db, remoteauth +from paskia.bootstrap import bootstrap_if_needed +from paskia.db import start_background from paskia.sansio import Passkey T = TypeVar("T") @@ -42,7 +45,6 @@ async def init( Set PASKIA_DB environment variable to specify the JSONL database file path. Default: paskia.jsonl """ - from . import db, remoteauth # Initialize passkey instance with provided parameters passkey.instance = Passkey( @@ -59,12 +61,10 @@ async def init( if bootstrap: # Bootstrap system if needed - from .bootstrap import bootstrap_if_needed await bootstrap_if_needed() # Start background flush/cleanup task after bootstrap - from .db import start_background await start_background() diff --git a/paskia/migrate/__init__.py b/paskia/migrate/__init__.py index 08b02c0..c9a9162 100644 --- a/paskia/migrate/__init__.py +++ b/paskia/migrate/__init__.py @@ -11,6 +11,7 @@ Or via the CLI entry point (if installed): paskia-migrate --sql sqlite+aiosqlite:///paskia.sqlite --json paskia.jsonl """ +import argparse import asyncio import re from datetime import datetime, timezone @@ -257,7 +258,6 @@ async def migrate_from_sql( def main(): """CLI entry point for migration.""" - import argparse parser = argparse.ArgumentParser( description="Migrate Paskia database from SQL to JSON" diff --git a/paskia/remoteauth.py b/paskia/remoteauth.py index d65d2bf..b2ec544 100644 --- a/paskia/remoteauth.py +++ b/paskia/remoteauth.py @@ -24,7 +24,7 @@ from datetime import datetime, timedelta, timezone from typing import Callable from uuid import UUID -from paskia.util import passphrase +from paskia.util import passphrase, pow # Remote auth requests expire after this duration REMOTE_AUTH_LIFETIME = timedelta(minutes=5) @@ -319,7 +319,6 @@ class RemoteAuthManager: Returns: PoW work units (pow.NORMAL or pow.HARD) """ - from paskia.util import pow count = self.get_connection_count() return pow.HARD if count >= 10 else pow.NORMAL diff --git a/paskia/util/hostutil.py b/paskia/util/hostutil.py index 1a4fa1b..0a6aa08 100644 --- a/paskia/util/hostutil.py +++ b/paskia/util/hostutil.py @@ -3,7 +3,7 @@ import json import os from functools import lru_cache -from urllib.parse import urlsplit +from urllib.parse import urlparse, urlsplit @lru_cache(maxsize=1) @@ -24,7 +24,6 @@ def dedicated_auth_host() -> str | None: auth_host = _load_config().get("auth_host") if not auth_host: return None - from urllib.parse import urlparse parsed = urlparse(auth_host if "://" in auth_host else f"//{auth_host}") return parsed.netloc or parsed.path or None diff --git a/paskia/util/vitedev.py b/paskia/util/vitedev.py index cd890b0..f4b34a0 100644 --- a/paskia/util/vitedev.py +++ b/paskia/util/vitedev.py @@ -10,6 +10,7 @@ but doesn't provide server-side fetching of HTML content. import asyncio import mimetypes import os +from importlib import resources from pathlib import Path import httpx @@ -24,7 +25,6 @@ def _get_dev_server() -> str | None: def _resolve_static_dir() -> Path: """Resolve the static files directory.""" - from importlib import resources # Try packaged path via importlib.resources (works for wheel/installed). try: # pragma: no cover - trivial path resolution diff --git a/pyproject.toml b/pyproject.toml index 8bb9e73..4c99000 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -79,7 +79,7 @@ target-version = "py39" line-length = 88 [tool.ruff.lint] -select = ["E", "F", "I", "N", "W", "UP"] +select = ["E", "F", "I", "N", "W", "UP", "PLC0415"] ignore = ["E501"] # Line too long isort.known-first-party = ["paskia"] diff --git a/tests/conftest.py b/tests/conftest.py index b8402b6..8094205 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -19,8 +19,9 @@ import httpx import pytest import pytest_asyncio +import paskia.db.operations as ops_db from paskia import globals as paskia_globals -from paskia.authsession import expires +from paskia.authsession import expires, reset_expires from paskia.db import ( Credential, Org, @@ -36,9 +37,12 @@ from paskia.db import ( create_session, create_user, ) +from paskia.db.jsonl import JsonlStore from paskia.db.operations import DB, _create_token +from paskia.fastapi.mainapp import app from paskia.fastapi.session import AUTH_COOKIE_NAME from paskia.sansio import Passkey +from paskia.util.passphrase import generate @pytest.fixture(scope="session") @@ -52,8 +56,6 @@ def event_loop(): @pytest_asyncio.fixture(scope="function") async def test_db() -> AsyncGenerator[DB, None]: """Create an in-memory JSON database for testing.""" - import paskia.db.operations as ops_db - from paskia.db.jsonl import JsonlStore with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=True) as f: db = DB() @@ -225,8 +227,6 @@ async def regular_session_token( @pytest_asyncio.fixture(scope="function") async def reset_token(test_db: DB, test_user: User, test_credential: Credential) -> str: """Create a reset token for the test user.""" - from paskia.authsession import reset_expires - from paskia.util.passphrase import generate token = generate() create_reset_token( @@ -248,7 +248,6 @@ async def client( initialized first. """ # Import app after globals are set - from paskia.fastapi.mainapp import app transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient( diff --git a/tests/test_admin.py b/tests/test_admin.py index 57a5ba4..b1e7fe2 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -11,6 +11,7 @@ These tests cover: - Credential management """ +import os from datetime import datetime, timezone from uuid import UUID @@ -19,6 +20,7 @@ import pytest import pytest_asyncio import uuid7 +from paskia import db from paskia.authsession import expires from paskia.db import ( Credential, @@ -78,7 +80,6 @@ async def second_org_user(test_db: DB, second_org_role: Role) -> User: @pytest_asyncio.fixture(scope="function") async def second_org_credential(test_db: DB, second_org_user: User) -> Credential: """Create a credential for the second org user.""" - import os credential = Credential.create( credential_id=os.urandom(32), @@ -139,7 +140,6 @@ async def org_admin_user(test_db: DB, org_admin_role: Role) -> User: @pytest_asyncio.fixture(scope="function") async def org_admin_credential(test_db: DB, org_admin_user: User) -> Credential: """Create a credential for the org admin user.""" - import os credential = Credential.create( credential_id=os.urandom(32), @@ -1423,7 +1423,6 @@ class TestAdminPermissions: ): """Cannot rename the auth:admin permission.""" # Get the auth:admin permission - from paskia import db perms = list(db.data().permissions.values()) admin_perm = next(p for p in perms if p.scope == "auth:admin") @@ -1478,7 +1477,6 @@ class TestAdminPermissions: ): """Cannot delete the only auth:admin permission (would lock out admin).""" # Get the auth:admin permission - from paskia import db perms = list(db.data().permissions.values()) admin_perm = next(p for p in perms if p.scope == "auth:admin") @@ -1496,14 +1494,12 @@ class TestAdminPermissions: self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Can delete an auth:admin permission if another accessible one exists.""" - from paskia.db import Permission # Create a second auth:admin permission (no domain restriction) perm2 = Permission.create(scope="auth:admin", display_name="Secondary Admin") create_permission(perm2) # Get the original auth:admin permission (the one created in setup) - from paskia import db perms = list(db.data().permissions.values()) admin_perms = [p for p in perms if p.scope == "auth:admin"] @@ -1524,7 +1520,6 @@ class TestAdminPermissions: self, client: httpx.AsyncClient, session_token: str, test_db: DB ): """Cannot delete auth:admin if remaining one has mismatched domain.""" - from paskia.db import Permission # Create a second auth:admin permission with a different domain perm2 = Permission.create( @@ -1536,7 +1531,6 @@ class TestAdminPermissions: # Cannot delete the original one because the remaining one is not accessible # Get the original auth:admin permission - from paskia import db perms = list(db.data().permissions.values()) admin_perms = [p for p in perms if p.scope == "auth:admin" and p.domain is None] diff --git a/tests/test_api.py b/tests/test_api.py index e79a482..b18632b 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -10,12 +10,15 @@ These tests cover: - /auth/api/set-session - Set session from bearer token """ -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone import httpx import pytest +from paskia.authsession import EXPIRES from paskia.db import create_session, delete_session +from paskia.db.operations import _create_token +from paskia.util.passphrase import generate from tests.conftest import auth_headers @@ -410,7 +413,6 @@ class TestTokenInfoEndpoint: @pytest.mark.asyncio async def test_token_info_with_nonexistent_token(self, client: httpx.AsyncClient): """Token info with well-formed but non-existent token should return 401.""" - from paskia.util.passphrase import generate fake_token = generate() response = await client.get( @@ -499,7 +501,6 @@ class TestValidateSessionRefresh: self, client: httpx.AsyncClient, test_db ): """Validate should handle session expiry during refresh attempt.""" - from paskia.db.operations import _create_token # Create a token but don't create a session for it token = _create_token() @@ -519,10 +520,6 @@ class TestValidateSessionRefresh: test_credential, ): """Validate should return 401 if session disappears during refresh.""" - from datetime import timedelta - - from paskia.authsession import EXPIRES - from paskia.db.operations import _create_token # Create a session with an old expiry time to trigger refresh token = _create_token()