diff --git a/Makefile b/Makefile index e1b9db5a..ccd27e0d 100644 --- a/Makefile +++ b/Makefile @@ -77,7 +77,7 @@ loadtest: ## Run locust against a server already on $(LOCUST_H loadtest-memray: ## Start uvicorn under memray, load-test, emit flamegraph scripts/loadtest_memray.sh $(LOCUST_ARGS) -lint: ci-python-lint ci-python-typecheck ci-js-lint ci-js-typecheck ci-check-file-size +lint: ci-python-lint ci-python-typecheck ci-js-lint ci-js-typecheck ci-check-file-size ci-check-hardcoded-strings # Kept granular so pr.yml can run them in parallel. ci-python-lint: @@ -103,6 +103,11 @@ ci-js-typecheck: ci-check-file-size: uv run python scripts/check_file_size.py +# Enforce that permissions, role names, Inertia page ids, and module dependency +# names are declared as named constants rather than hardcoded string literals. +ci-check-hardcoded-strings: + uv run python scripts/check_hardcoded_strings.py + # Diagnostics doctor: uv run python -m simple_module_core diff --git a/framework/db/simple_module_db/listeners.py b/framework/db/simple_module_db/listeners.py index 52d8a867..88a49a66 100644 --- a/framework/db/simple_module_db/listeners.py +++ b/framework/db/simple_module_db/listeners.py @@ -32,6 +32,18 @@ class TenantIsolationError(Exception): # flush has cleared ``session.new/.dirty/.deleted``. SESSION_HAS_WRITES_KEY = "has_writes" +# DB audit event names +_EVENT_ENTITY_CREATED = "db.entity.created" +_EVENT_ENTITY_UPDATED = "db.entity.updated" +_EVENT_ENTITY_SOFT_DELETED = "db.entity.soft_deleted" +_EVENT_ENTITY_DELETED = "db.entity.deleted" + +# DB operation strings used in log extra dicts +_OP_CREATE = "create" +_OP_UPDATE = "update" +_OP_SOFT_DELETE = "soft_delete" +_OP_DELETE = "delete" + def _entity_label(obj: object) -> str: """Return 'ClassName' for a mapped entity instance.""" @@ -107,9 +119,9 @@ def _before_flush_listener( ) _db_logger.info( - "db.entity.created", + _EVENT_ENTITY_CREATED, extra={ - "operation": "create", + "operation": _OP_CREATE, "entity": _entity_label(obj), "user_id": user_id, }, @@ -133,9 +145,9 @@ def _before_flush_listener( raise TenantIsolationError("Cannot change tenant_id of an existing object") _db_logger.info( - "db.entity.updated", + _EVENT_ENTITY_UPDATED, extra={ - "operation": "update", + "operation": _OP_UPDATE, "entity": _entity_label(obj), "entity_id": _entity_pk(obj), "user_id": user_id, @@ -154,9 +166,9 @@ def _before_flush_listener( session.add(obj) _db_logger.info( - "db.entity.soft_deleted", + _EVENT_ENTITY_SOFT_DELETED, extra={ - "operation": "soft_delete", + "operation": _OP_SOFT_DELETE, "entity": _entity_label(obj), "entity_id": _entity_pk(obj), "user_id": user_id, @@ -164,9 +176,9 @@ def _before_flush_listener( ) else: _db_logger.info( - "db.entity.deleted", + _EVENT_ENTITY_DELETED, extra={ - "operation": "delete", + "operation": _OP_DELETE, "entity": _entity_label(obj), "entity_id": _entity_pk(obj), "user_id": user_id, diff --git a/framework/hosting/simple_module_hosting/_inertia_setup.py b/framework/hosting/simple_module_hosting/_inertia_setup.py index 1c66fcc2..9dcd10d4 100644 --- a/framework/hosting/simple_module_hosting/_inertia_setup.py +++ b/framework/hosting/simple_module_hosting/_inertia_setup.py @@ -12,6 +12,11 @@ logger = logging.getLogger(__name__) +_INERTIA_VERSION = "1.0" +_ROOT_TEMPLATE_FILENAME = "index.html" +_ENTRYPOINT_FILENAME = "main.tsx" +_ROOT_DIRECTORY = "." + def setup_inertia( app: FastAPI, @@ -54,12 +59,12 @@ def setup_inertia( inertia_config = InertiaConfig( environment=settings.environment, - version="1.0", + version=_INERTIA_VERSION, dev_url=settings.vite_dev_url if settings.is_development else "", templates=templates, - root_template_filename="index.html", - entrypoint_filename="main.tsx", - root_directory=".", + root_template_filename=_ROOT_TEMPLATE_FILENAME, + entrypoint_filename=_ENTRYPOINT_FILENAME, + root_directory=_ROOT_DIRECTORY, use_flash_errors=True, ) diff --git a/framework/hosting/simple_module_hosting/_inertia_shared.py b/framework/hosting/simple_module_hosting/_inertia_shared.py index 654a662e..bd3522f1 100644 --- a/framework/hosting/simple_module_hosting/_inertia_shared.py +++ b/framework/hosting/simple_module_hosting/_inertia_shared.py @@ -11,6 +11,8 @@ logger = logging.getLogger(__name__) _I18N_SESSION_LOCALE_KEY = "__i18n_locale" +_INERTIA_HEADER = "x-inertia" +_INERTIA_HEADER_TRUE = "true" def build_i18n_block(scope: Scope, request: Request) -> dict: @@ -38,7 +40,7 @@ def build_i18n_block(scope: Scope, request: Request) -> dict: ) return {"locale": "en", "supportedLocales": ["en"], "messages": {}} - is_inertia = Headers(scope=scope).get("x-inertia") == "true" + is_inertia = Headers(scope=scope).get(_INERTIA_HEADER) == _INERTIA_HEADER_TRUE session_dict = scope.get("session") # When the session is absent (pre-session-middleware routes, WebSocket # upgrades), treat locale as "unchanged" so Inertia XHR requests still diff --git a/framework/hosting/simple_module_hosting/_observability.py b/framework/hosting/simple_module_hosting/_observability.py index 66ba9d6a..c6a7353d 100644 --- a/framework/hosting/simple_module_hosting/_observability.py +++ b/framework/hosting/simple_module_hosting/_observability.py @@ -11,7 +11,13 @@ from simple_module_hosting.logging import correlation_id -_request_logger = logging.getLogger("simple_module.request") +_LOGGER_NAME = "simple_module.request" +_request_logger = logging.getLogger(_LOGGER_NAME) + +_SCOPE_HTTP = "http" +_MSG_RESPONSE_START = "http.response.start" +_EVENT_REQUEST_STARTED = "request.started" +_EVENT_REQUEST_COMPLETED = "request.completed" # Paths that produce noisy, low-value log entries _QUIET_PREFIXES = ("/health", "/static/") @@ -32,14 +38,14 @@ def __init__(self, app: ASGIApp) -> None: self.app = app async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": + if scope["type"] != _SCOPE_HTTP: await self.app(scope, receive, send) return cid = Headers(scope=scope).get(self.HEADER) or uuid.uuid4().hex async def send_with_header(message: Message) -> None: - if message["type"] == "http.response.start": + if message["type"] == _MSG_RESPONSE_START: headers = MutableHeaders(scope=message) headers[self.HEADER] = cid await send(message) @@ -58,7 +64,7 @@ def __init__(self, app: ASGIApp) -> None: self.app = app async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": + if scope["type"] != _SCOPE_HTTP: await self.app(scope, receive, send) return @@ -72,7 +78,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: client_ip = client[0] if client else "unknown" _request_logger.debug( - "request.started", + _EVENT_REQUEST_STARTED, extra={"method": method, "path": path, "client_ip": client_ip}, ) @@ -81,7 +87,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: async def send_capture(message: Message) -> None: nonlocal status_code - if message["type"] == "http.response.start": + if message["type"] == _MSG_RESPONSE_START: status_code = message["status"] await send(message) @@ -91,7 +97,7 @@ async def send_capture(message: Message) -> None: # Log completion even when the inner app raises, so 500s are observable. duration_ms = round((time.perf_counter() - start) * 1000, 2) _request_logger.info( - "request.completed", + _EVENT_REQUEST_COMPLETED, extra={ "method": method, "path": path, diff --git a/framework/hosting/simple_module_hosting/app_builder.py b/framework/hosting/simple_module_hosting/app_builder.py index a75300ae..b00b4f52 100644 --- a/framework/hosting/simple_module_hosting/app_builder.py +++ b/framework/hosting/simple_module_hosting/app_builder.py @@ -35,6 +35,14 @@ logger = logging.getLogger(__name__) +_APP_TITLE = "SimpleModule" +_APP_VERSION = "0.1.0" +_DOCS_URL = "/api/docs" +_REDOC_URL = "/api/redoc" +_STATIC_MOUNT_PATH = "/static" +_STATIC_DIR_NAME = "static" +_ENV_PROJECT_ROOT = "SM_PROJECT_ROOT" + def _resolve_project_root() -> Path: """Return the project root directory. @@ -48,7 +56,7 @@ def _resolve_project_root() -> Path: Falls back to ``parents[3]`` for the workspace-install dev loop (simple_module_hosting/ → hosting/ → framework/ → project root). """ - override = os.environ.get("SM_PROJECT_ROOT") + override = os.environ.get(_ENV_PROJECT_ROOT) if override: return Path(override) return Path(__file__).resolve().parents[3] @@ -152,10 +160,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: await app.state.sm.db.engine.dispose() app = FastAPI( - title="SimpleModule", - version="0.1.0", - docs_url="/api/docs" if settings.is_development else None, - redoc_url="/api/redoc" if settings.is_development else None, + title=_APP_TITLE, + version=_APP_VERSION, + docs_url=_DOCS_URL if settings.is_development else None, + redoc_url=_REDOC_URL if settings.is_development else None, lifespan=lifespan, ) @@ -210,9 +218,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: app.include_router(health_router) - static_dir = _PROJECT_ROOT / "host" / "static" + static_dir = _PROJECT_ROOT / "host" / _STATIC_DIR_NAME if static_dir.is_dir(): - app.mount("/static", StaticFiles(directory=static_dir), name="static") + app.mount(_STATIC_MOUNT_PATH, StaticFiles(directory=static_dir), name=_STATIC_DIR_NAME) mount_module_static_dirs(app, modules) diff --git a/framework/hosting/simple_module_hosting/csrf.py b/framework/hosting/simple_module_hosting/csrf.py index a151d148..a284a241 100644 --- a/framework/hosting/simple_module_hosting/csrf.py +++ b/framework/hosting/simple_module_hosting/csrf.py @@ -23,8 +23,11 @@ """Session-dict key under which :class:`InertiaLayoutDataMiddleware` mints the per-session CSRF token that :class:`CSRFMiddleware` validates.""" +_SCOPE_HTTP = "http" _SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE"}) _HEADER_NAMES = ("x-csrf-token", "x-xsrf-token") +_EVENT_CSRF_REJECTED = "csrf.rejected" +_CSRF_VALIDATION_FAILED_MSG = "CSRF validation failed" class CSRFMiddleware: @@ -45,7 +48,7 @@ def __init__( self.exempt_path_prefixes = exempt_path_prefixes async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": + if scope["type"] != _SCOPE_HTTP: await self.app(scope, receive, send) return @@ -72,10 +75,10 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if not expected or not provided or not secrets.compare_digest(str(expected), provided): logger.info( - "csrf.rejected", + _EVENT_CSRF_REJECTED, extra={"method": method, "path": path, "has_token": bool(expected)}, ) - response = PlainTextResponse("CSRF validation failed", status_code=403) + response = PlainTextResponse(_CSRF_VALIDATION_FAILED_MSG, status_code=403) await response(scope, receive, send) return diff --git a/framework/hosting/simple_module_hosting/health.py b/framework/hosting/simple_module_hosting/health.py index a91286bc..3abc047a 100644 --- a/framework/hosting/simple_module_hosting/health.py +++ b/framework/hosting/simple_module_hosting/health.py @@ -12,6 +12,13 @@ router = APIRouter(tags=["Health"]) +_KEY_STATUS = "status" +_KEY_CHECKS = "checks" +_KEY_MIGRATION = "migration" +_KEY_DETAIL = "detail" +_STATUS_HEALTHY = "healthy" +_STATUS_ALIVE = "alive" + # Severity ordering for aggregation: worst status wins _STATUS_SEVERITY = { HealthStatus.HEALTHY: 0, @@ -24,14 +31,14 @@ async def health(request: Request) -> dict: migration = getattr(request.app.state, "migration", None) return { - "status": "healthy", - "migration": migration, + _KEY_STATUS: _STATUS_HEALTHY, + _KEY_MIGRATION: migration, } @router.get("/health/live") async def liveness() -> dict: - return {"status": "alive"} + return {_KEY_STATUS: _STATUS_ALIVE} @router.get("/health/ready") @@ -40,7 +47,7 @@ async def readiness(request: Request) -> dict: checks = registry.all_checks if not checks: - return {"status": "healthy", "checks": {}} + return {_KEY_STATUS: _STATUS_HEALTHY, _KEY_CHECKS: {}} # Run all checks concurrently async def _run_check(name: str, check_fn): @@ -64,9 +71,9 @@ async def _run_check(name: str, check_fn): worst = result.status return { - "status": worst.value, - "checks": { - name: {"status": r.status.value, **({"detail": r.detail} if r.detail else {})} + _KEY_STATUS: worst.value, + _KEY_CHECKS: { + name: {_KEY_STATUS: r.status.value, **({_KEY_DETAIL: r.detail} if r.detail else {})} for name, r in results.items() }, } diff --git a/framework/hosting/simple_module_hosting/middleware.py b/framework/hosting/simple_module_hosting/middleware.py index 2989b9f2..3b944910 100644 --- a/framework/hosting/simple_module_hosting/middleware.py +++ b/framework/hosting/simple_module_hosting/middleware.py @@ -32,6 +32,23 @@ logger = logging.getLogger(__name__) +_SCOPE_HTTP = "http" +_MSG_RESPONSE_START = "http.response.start" + +# Security response header names +_HEADER_X_CONTENT_TYPE_OPTIONS = "X-Content-Type-Options" +_HEADER_X_FRAME_OPTIONS = "X-Frame-Options" +_HEADER_X_XSS_PROTECTION = "X-XSS-Protection" +_HEADER_REFERRER_POLICY = "Referrer-Policy" +_HEADER_CSP = "Content-Security-Policy" +_HEADER_HSTS = "Strict-Transport-Security" + +# Security response header values +_XCTO_NOSNIFF = "nosniff" +_XFO_SAMEORIGIN = "SAMEORIGIN" +_XXSS_BLOCK = "1; mode=block" +_REFERRER_STRICT_ORIGIN = "strict-origin-when-cross-origin" + __all__ = [ "TENANT_HEADER", "CorrelationIdMiddleware", @@ -105,21 +122,21 @@ def __init__( self.hsts = strict_transport_security async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": + if scope["type"] != _SCOPE_HTTP: await self.app(scope, receive, send) return async def send_with_headers(message: Message) -> None: - if message["type"] == "http.response.start": + if message["type"] == _MSG_RESPONSE_START: headers = MutableHeaders(scope=message) - headers["X-Content-Type-Options"] = "nosniff" - headers["X-Frame-Options"] = "SAMEORIGIN" - headers["X-XSS-Protection"] = "1; mode=block" - headers["Referrer-Policy"] = "strict-origin-when-cross-origin" + headers[_HEADER_X_CONTENT_TYPE_OPTIONS] = _XCTO_NOSNIFF + headers[_HEADER_X_FRAME_OPTIONS] = _XFO_SAMEORIGIN + headers[_HEADER_X_XSS_PROTECTION] = _XXSS_BLOCK + headers[_HEADER_REFERRER_POLICY] = _REFERRER_STRICT_ORIGIN if self.csp: - headers["Content-Security-Policy"] = self.csp + headers[_HEADER_CSP] = self.csp if self.hsts: - headers["Strict-Transport-Security"] = self.hsts + headers[_HEADER_HSTS] = self.hsts await send(message) await self.app(scope, receive, send_with_headers) @@ -151,7 +168,7 @@ def __init__(self, app: ASGIApp, *, header: str | None = None) -> None: self.header = header async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": + if scope["type"] != _SCOPE_HTTP: await self.app(scope, receive, send) return @@ -198,7 +215,7 @@ def __init__( self.permission_registry = permission_registry async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": + if scope["type"] != _SCOPE_HTTP: await self.app(scope, receive, send) return diff --git a/modules/auth/auth/deps.py b/modules/auth/auth/deps.py index c1fa26c9..7bac3e0f 100644 --- a/modules/auth/auth/deps.py +++ b/modules/auth/auth/deps.py @@ -9,6 +9,8 @@ from auth.contracts.schemas import UserContext +_ADMIN_ROLE = "admin" + async def get_current_user(request: Request, t: TranslatorDep) -> UserContext: """Extract the authenticated user from request state. @@ -39,7 +41,7 @@ async def check( user: UserContext = Depends(get_current_user), ): # Admin role bypasses permission checks - if "admin" in user.roles: + if _ADMIN_ROLE in user.roles: return # Get permission registry from app state diff --git a/modules/dashboard/dashboard/module.py b/modules/dashboard/dashboard/module.py index 17fa7070..05df1763 100644 --- a/modules/dashboard/dashboard/module.py +++ b/modules/dashboard/dashboard/module.py @@ -9,13 +9,18 @@ from simple_module_core.menu import MenuItem, MenuRegistry, MenuSection from simple_module_core.module import ModuleBase, ModuleMeta +_MODULE_PRODUCTS = "Products" +_MODULE_USERS = "Users" +_URL_DASHBOARD = "/dashboard" +_ICON_DASHBOARD = "home" + class DashboardModule(ModuleBase): meta = ModuleMeta( name="Dashboard", route_prefix="/api/dashboard", view_prefix="/dashboard", - depends_on=["Products", "Users"], + depends_on=[_MODULE_PRODUCTS, _MODULE_USERS], ) def register_routes(self, api_router: APIRouter, view_router: APIRouter) -> None: @@ -29,8 +34,8 @@ def register_menu_items(self, registry: MenuRegistry) -> None: registry.add( MenuItem( label="Dashboard", - url="/dashboard", - icon="home", + url=_URL_DASHBOARD, + icon=_ICON_DASHBOARD, order=1, section=MenuSection.SIDEBAR, ) diff --git a/modules/products/products/constants.py b/modules/products/products/constants.py new file mode 100644 index 00000000..229c0a43 --- /dev/null +++ b/modules/products/products/constants.py @@ -0,0 +1,10 @@ +"""Stable identifiers for the products module.""" + +# Permission identifiers +PERM_PRODUCTS_VIEW = "products.view" +PERM_PRODUCTS_CREATE = "products.create" +PERM_PRODUCTS_EDIT = "products.edit" +PERM_PRODUCTS_DELETE = "products.delete" + +# Feature flag identifiers +FLAG_PRODUCTS_BULK_IMPORT = "products.bulk_import" diff --git a/modules/products/products/endpoints/api.py b/modules/products/products/endpoints/api.py index 2cf3acb6..ed619619 100644 --- a/modules/products/products/endpoints/api.py +++ b/modules/products/products/endpoints/api.py @@ -7,6 +7,7 @@ from simple_module_hosting.i18n_deps import TranslatorDep from simple_module_hosting.permissions import RequiresPermission +from products.constants import PERM_PRODUCTS_CREATE, PERM_PRODUCTS_DELETE, PERM_PRODUCTS_EDIT from products.contracts.events import ProductCreated, ProductDeleted, ProductUpdated from products.contracts.schemas import ProductCreate, ProductOut, ProductUpdate from products.deps import get_event_bus, get_product_service @@ -39,7 +40,7 @@ async def get_product( "/", response_model=ProductOut, status_code=201, - dependencies=[Depends(RequiresPermission("products.create"))], + dependencies=[Depends(RequiresPermission(PERM_PRODUCTS_CREATE))], ) async def create_product( data: ProductCreate, @@ -54,7 +55,7 @@ async def create_product( @router.put( "/{product_id}", response_model=ProductOut, - dependencies=[Depends(RequiresPermission("products.edit"))], + dependencies=[Depends(RequiresPermission(PERM_PRODUCTS_EDIT))], ) async def update_product( product_id: int, @@ -73,7 +74,7 @@ async def update_product( @router.delete( "/{product_id}", status_code=204, - dependencies=[Depends(RequiresPermission("products.delete"))], + dependencies=[Depends(RequiresPermission(PERM_PRODUCTS_DELETE))], ) async def delete_product( product_id: int, diff --git a/modules/products/products/endpoints/views.py b/modules/products/products/endpoints/views.py index c4dccc63..e868fd94 100644 --- a/modules/products/products/endpoints/views.py +++ b/modules/products/products/endpoints/views.py @@ -11,6 +11,7 @@ from simple_module_hosting.permissions import RequiresPermission from starlette.responses import RedirectResponse +from products.constants import PERM_PRODUCTS_CREATE, PERM_PRODUCTS_DELETE, PERM_PRODUCTS_EDIT from products.contracts.schemas import ProductCreate, ProductUpdate from products.deps import get_product_service from products.service import ProductService @@ -18,6 +19,12 @@ router = APIRouter() PER_PAGE = 10 +_REDIRECT_PRODUCTS = "/products" + +# Inertia page identifiers +_PAGE_BROWSE = "Products/Browse" +_PAGE_CREATE = "Products/Create" +_PAGE_EDIT = "Products/Edit" # ── View routes (GET → Inertia pages) ───────────────────────── @@ -36,7 +43,7 @@ async def browse( search=search or None, ) return await inertia.render( - "Products/Browse", + _PAGE_BROWSE, { "products": [p.model_dump(mode="json") for p in products], "pagination": { @@ -52,10 +59,10 @@ async def browse( @router.get( "/create", response_model=None, - dependencies=[Depends(RequiresPermission("products.create"))], + dependencies=[Depends(RequiresPermission(PERM_PRODUCTS_CREATE))], ) async def create_view(inertia: InertiaDep) -> InertiaResponse: - return await inertia.render("Products/Create") + return await inertia.render(_PAGE_CREATE) @router.get("/{product_id}/edit", response_model=None) @@ -67,9 +74,9 @@ async def edit_view( ) -> InertiaResponse: product = await service.get_by_id(product_id) if product is None: - return await inertia.render("Products/Browse", {"error": t.t("products.errors.not_found")}) + return await inertia.render(_PAGE_BROWSE, {"error": t.t("products.errors.not_found")}) return await inertia.render( - "Products/Edit", + _PAGE_EDIT, {"product": product.model_dump(mode="json")}, ) @@ -80,7 +87,7 @@ async def edit_view( @router.post( "/", response_model=None, - dependencies=[Depends(RequiresPermission("products.create"))], + dependencies=[Depends(RequiresPermission(PERM_PRODUCTS_CREATE))], ) async def create_action( request: Request, @@ -92,13 +99,13 @@ async def create_action( except ValidationError as exc: return redirect_back_with_errors(request, validation_errors_to_dict(exc)) await service.create(data) - return RedirectResponse("/products", status_code=303) + return RedirectResponse(_REDIRECT_PRODUCTS, status_code=303) @router.put( "/{product_id}", response_model=None, - dependencies=[Depends(RequiresPermission("products.edit"))], + dependencies=[Depends(RequiresPermission(PERM_PRODUCTS_EDIT))], ) async def update_action( product_id: int, @@ -111,17 +118,17 @@ async def update_action( except ValidationError as exc: return redirect_back_with_errors(request, validation_errors_to_dict(exc)) await service.update(product_id, data) - return RedirectResponse("/products", status_code=303) + return RedirectResponse(_REDIRECT_PRODUCTS, status_code=303) @router.delete( "/{product_id}", response_model=None, - dependencies=[Depends(RequiresPermission("products.delete"))], + dependencies=[Depends(RequiresPermission(PERM_PRODUCTS_DELETE))], ) async def delete_action( product_id: int, service: ProductService = Depends(get_product_service), ) -> RedirectResponse: await service.delete(product_id) - return RedirectResponse("/products", status_code=303) + return RedirectResponse(_REDIRECT_PRODUCTS, status_code=303) diff --git a/modules/products/products/module.py b/modules/products/products/module.py index 2addfd0c..be88a4d9 100644 --- a/modules/products/products/module.py +++ b/modules/products/products/module.py @@ -11,6 +11,17 @@ from simple_module_core.module import ModuleBase, ModuleMeta from simple_module_core.permissions import PermissionRegistry +from products.constants import ( + FLAG_PRODUCTS_BULK_IMPORT, + PERM_PRODUCTS_CREATE, + PERM_PRODUCTS_DELETE, + PERM_PRODUCTS_EDIT, + PERM_PRODUCTS_VIEW, +) + +_URL_PRODUCTS = "/products" +_ICON_PRODUCTS = "package" + class ProductsModule(ModuleBase): meta = ModuleMeta( @@ -30,8 +41,8 @@ def register_menu_items(self, registry: MenuRegistry) -> None: registry.add( MenuItem( label="Products", - url="/products", - icon="package", + url=_URL_PRODUCTS, + icon=_ICON_PRODUCTS, order=20, section=MenuSection.SIDEBAR, ) @@ -41,17 +52,17 @@ def register_permissions(self, registry: PermissionRegistry) -> None: registry.add_group( "Products", [ - "products.view", - "products.create", - "products.edit", - "products.delete", + PERM_PRODUCTS_VIEW, + PERM_PRODUCTS_CREATE, + PERM_PRODUCTS_EDIT, + PERM_PRODUCTS_DELETE, ], ) def register_feature_flags(self, registry: FeatureFlagRegistry) -> None: registry.add( FeatureFlagDefinition( - name="products.bulk_import", + name=FLAG_PRODUCTS_BULK_IMPORT, description="Enable bulk product import from CSV", default_enabled=False, ) diff --git a/modules/users/users/backend.py b/modules/users/users/backend.py index 55bb7e93..66059cf5 100644 --- a/modules/users/users/backend.py +++ b/modules/users/users/backend.py @@ -28,6 +28,9 @@ if TYPE_CHECKING: from users.settings import UsersSettings +_TOKEN_LIFETIME_SECONDS = 60 * 60 * 24 * 14 # 14 days +_AUTH_BACKEND_NAME = "cookie" + def build_cookie_transport( cookie_name: str, @@ -46,7 +49,7 @@ def build_cookie_transport( def get_database_strategy( access_token_db: AccessTokenDatabase[UserAccessToken] = Depends(get_access_token_db), - lifetime_seconds: int = 60 * 60 * 24 * 14, + lifetime_seconds: int = _TOKEN_LIFETIME_SECONDS, ) -> DatabaseStrategy: return DatabaseStrategy(access_token_db, lifetime_seconds=lifetime_seconds) @@ -55,7 +58,7 @@ def build_auth_backend( cookie_transport: CookieTransport, ) -> AuthenticationBackend: return AuthenticationBackend( - name="cookie", + name=_AUTH_BACKEND_NAME, transport=cookie_transport, get_strategy=get_database_strategy, ) diff --git a/modules/users/users/bootstrap.py b/modules/users/users/bootstrap.py index 2bad9aa8..516a7014 100644 --- a/modules/users/users/bootstrap.py +++ b/modules/users/users/bootstrap.py @@ -9,12 +9,26 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession -from users.constants import ADMIN_ROLE_ID, USER_ROLE_ID +from users.constants import ( + ADMIN_ROLE_DESCRIPTION, + ADMIN_ROLE_ID, + ADMIN_ROLE_NAME, + USER_ROLE_DESCRIPTION, + USER_ROLE_ID, + USER_ROLE_NAME, +) from users.models import Role, User, UserRole from users.settings import UsersSettings logger = logging.getLogger("users.bootstrap") +_EVT_CREATED = "users.bootstrap.created" +_EVT_UPDATED = "users.bootstrap.updated" +_EVT_NOOP = "users.bootstrap.noop" +_EVT_USER_NOOP = "users.bootstrap.user_noop" +_EVT_USER_CREATED = "users.bootstrap.user_created" +_EVT_FAILED = "users.bootstrap.failed" + @dataclass class CreateAdminResult: @@ -51,7 +65,9 @@ async def create_admin( # Look up by name first (works on both Postgres and SQLite regardless of UUID # storage format), then fall back to id-based lookup in case name was changed. - admin_role = (await db.execute(select(Role).where(Role.name == "admin"))).scalar_one_or_none() + admin_role = ( + await db.execute(select(Role).where(Role.name == ADMIN_ROLE_NAME)) + ).scalar_one_or_none() if admin_role is None: admin_role = ( await db.execute(select(Role).where(Role.id == ADMIN_ROLE_ID)) @@ -62,7 +78,9 @@ async def create_admin( # tests (where `create_all` runs without data migrations) and for # scenarios where someone ran `alembic downgrade` past the seed # revision but not past the schema revision. - admin_role = Role(id=ADMIN_ROLE_ID, name="admin", description="Administrator") + admin_role = Role( + id=ADMIN_ROLE_ID, name=ADMIN_ROLE_NAME, description=ADMIN_ROLE_DESCRIPTION + ) db.add(admin_role) await db.flush() @@ -81,7 +99,7 @@ async def create_admin( db.add(UserRole(user_id=user.id, role_id=admin_role.id)) await db.commit() await db.refresh(user) - logger.info("users.bootstrap.created", extra={"email": email, "id": str(user.id)}) + logger.info(_EVT_CREATED, extra={"email": email, "id": str(user.id)}) return CreateAdminResult(user=user, created=True) if force: @@ -101,13 +119,10 @@ async def create_admin( if existing_link is None: db.add(UserRole(user_id=existing.id, role_id=admin_role.id)) await db.commit() - logger.info( - "users.bootstrap.updated", - extra={"email": email, "id": str(existing.id)}, - ) + logger.info(_EVT_UPDATED, extra={"email": email, "id": str(existing.id)}) return CreateAdminResult(user=existing, created=False) - logger.info("users.bootstrap.noop", extra={"email": email, "id": str(existing.id)}) + logger.info(_EVT_NOOP, extra={"email": email, "id": str(existing.id)}) return CreateAdminResult(user=existing, created=False) @@ -129,17 +144,19 @@ async def create_standard_user( await db.execute(select(User).where(func.lower(User.email) == email.lower())) ).scalar_one_or_none() if existing is not None: - logger.info("users.bootstrap.user_noop", extra={"email": email, "id": str(existing.id)}) + logger.info(_EVT_USER_NOOP, extra={"email": email, "id": str(existing.id)}) return CreateAdminResult(user=existing, created=False) - user_role = (await db.execute(select(Role).where(Role.name == "user"))).scalar_one_or_none() + user_role = ( + await db.execute(select(Role).where(Role.name == USER_ROLE_NAME)) + ).scalar_one_or_none() if user_role is None: user_role = ( await db.execute(select(Role).where(Role.id == USER_ROLE_ID)) ).scalar_one_or_none() if user_role is None: # Safety net — the seed migration normally inserts this row. - user_role = Role(id=USER_ROLE_ID, name="user", description="Standard user") + user_role = Role(id=USER_ROLE_ID, name=USER_ROLE_NAME, description=USER_ROLE_DESCRIPTION) db.add(user_role) await db.flush() @@ -156,7 +173,7 @@ async def create_standard_user( db.add(UserRole(user_id=user.id, role_id=user_role.id)) await db.commit() await db.refresh(user) - logger.info("users.bootstrap.user_created", extra={"email": email, "id": str(user.id)}) + logger.info(_EVT_USER_CREATED, extra={"email": email, "id": str(user.id)}) return CreateAdminResult(user=user, created=True) @@ -199,5 +216,5 @@ async def bootstrap_admin_from_env(app: FastAPI) -> None: password=settings.bootstrap_user_password, ) except Exception: - logger.exception("users.bootstrap.failed") + logger.exception(_EVT_FAILED) raise diff --git a/modules/users/users/constants.py b/modules/users/users/constants.py index e0ae3834..7b04db1c 100644 --- a/modules/users/users/constants.py +++ b/modules/users/users/constants.py @@ -4,3 +4,18 @@ ADMIN_ROLE_ID = uuid.UUID("00000000-0000-0000-0000-000000000001") USER_ROLE_ID = uuid.UUID("00000000-0000-0000-0000-000000000002") + +# Role name strings +ADMIN_ROLE_NAME = "admin" +USER_ROLE_NAME = "user" + +# Role descriptions +ADMIN_ROLE_DESCRIPTION = "Administrator" +USER_ROLE_DESCRIPTION = "Standard user" + +# Permission identifiers +PERM_USERS_MANAGE = "users.manage" +PERM_USERS_SELF_PROFILE = "users.self.profile" + +# Session keys +SESSION_USER_ID_KEY = "user_id" diff --git a/modules/users/users/endpoints/api.py b/modules/users/users/endpoints/api.py index b0584168..049c4ceb 100644 --- a/modules/users/users/endpoints/api.py +++ b/modules/users/users/endpoints/api.py @@ -16,6 +16,7 @@ from fastapi.security import OAuth2PasswordRequestForm from fastapi_users import exceptions as fu_exceptions +from users.constants import SESSION_USER_ID_KEY from users.contracts.schemas import ( AcceptInviteRequest, SelfProfileUpdate, @@ -99,8 +100,8 @@ async def login( await user_manager.on_after_login(user, request, response) # Set fastapi-users' cookie via auth_backend.login login_response = await auth_backend.login(strategy, user) - # Bridge the session cookie — AuthMiddleware (once wired in Task 8) reads this - request.session["user_id"] = str(user.id) + # Bridge the session cookie — AuthMiddleware reads this to identify the user + request.session[SESSION_USER_ID_KEY] = str(user.id) return login_response @@ -177,7 +178,7 @@ async def accept_invite( await user_manager.on_after_login(user, request, response) login_response = await auth_backend.login(strategy, user) - request.session["user_id"] = str(user.id) + request.session[SESSION_USER_ID_KEY] = str(user.id) return login_response diff --git a/modules/users/users/endpoints/api_admin.py b/modules/users/users/endpoints/api_admin.py index 7e547be7..dd0849da 100644 --- a/modules/users/users/endpoints/api_admin.py +++ b/modules/users/users/endpoints/api_admin.py @@ -12,6 +12,7 @@ from simple_module_core.events import EventBus from simple_module_hosting.permissions import RequiresPermission +from users.constants import PERM_USERS_MANAGE from users.contracts.events import RoleAssigned, UserDisabled, UserInvited from users.contracts.schemas import ( PasswordResetLink, @@ -24,7 +25,7 @@ admin_router = APIRouter( prefix="/admin", - dependencies=[Depends(RequiresPermission("users.manage"))], + dependencies=[Depends(RequiresPermission(PERM_USERS_MANAGE))], tags=["users-admin"], ) diff --git a/modules/users/users/endpoints/views.py b/modules/users/users/endpoints/views.py index 72c92b4c..8b641a5d 100644 --- a/modules/users/users/endpoints/views.py +++ b/modules/users/users/endpoints/views.py @@ -10,12 +10,25 @@ from simple_module_hosting.permissions import RequiresPermission from starlette.responses import RedirectResponse +from users.constants import PERM_USERS_MANAGE from users.deps import get_user_service from users.roles_cache import get_roles_cache from users.service import UserService router = APIRouter() +# Inertia page identifiers +_PAGE_LOGIN = "Users/Login" +_PAGE_REGISTER = "Users/Register" +_PAGE_FORGOT_PASSWORD = "Users/ForgotPassword" +_PAGE_RESET_PASSWORD = "Users/ResetPassword" +_PAGE_VERIFY_EMAIL = "Users/VerifyEmail" +_PAGE_ACCEPT_INVITE = "Users/AcceptInvite" +_PAGE_PROFILE = "Users/Profile" +_PAGE_ADMIN_INDEX = "Users/Users/Index" +_PAGE_ADMIN_INVITE = "Users/Users/Invite" +_PAGE_ADMIN_EDIT = "Users/Users/Edit" + # ── Public auth pages ─────────────────────────────────────────── @@ -45,7 +58,7 @@ async def login_page(request: Request, inertia: InertiaDep) -> InertiaResponse: } ) return await inertia.render( - "Users/Login", + _PAGE_LOGIN, {"allow_signup": users_settings.allow_signup, "dev_accounts": dev_accounts}, ) @@ -67,27 +80,27 @@ async def logout(request: Request) -> RedirectResponse: async def register_page(request: Request, inertia: InertiaDep) -> InertiaResponse: if not request.app.state.users.settings.allow_signup: raise HTTPException(status_code=404) - return await inertia.render("Users/Register", {}) + return await inertia.render(_PAGE_REGISTER, {}) @router.get("/forgot-password", response_model=None) async def forgot_password_page(inertia: InertiaDep) -> InertiaResponse: - return await inertia.render("Users/ForgotPassword", {}) + return await inertia.render(_PAGE_FORGOT_PASSWORD, {}) @router.get("/reset-password", response_model=None) async def reset_password_page(inertia: InertiaDep, token: str = "") -> InertiaResponse: - return await inertia.render("Users/ResetPassword", {"token": token}) + return await inertia.render(_PAGE_RESET_PASSWORD, {"token": token}) @router.get("/verify", response_model=None) async def verify_page(inertia: InertiaDep, token: str = "") -> InertiaResponse: - return await inertia.render("Users/VerifyEmail", {"token": token}) + return await inertia.render(_PAGE_VERIFY_EMAIL, {"token": token}) @router.get("/invite/accept", response_model=None) async def accept_invite_page(inertia: InertiaDep, token: str = "") -> InertiaResponse: - return await inertia.render("Users/AcceptInvite", {"token": token}) + return await inertia.render(_PAGE_ACCEPT_INVITE, {"token": token}) # ── Authenticated pages ───────────────────────────────────────── @@ -95,7 +108,7 @@ async def accept_invite_page(inertia: InertiaDep, token: str = "") -> InertiaRes @router.get("/me", response_model=None) async def profile_page(inertia: InertiaDep) -> InertiaResponse: - return await inertia.render("Users/Profile", {}) + return await inertia.render(_PAGE_PROFILE, {}) # ── Admin pages ───────────────────────────────────────────────── @@ -104,7 +117,7 @@ async def profile_page(inertia: InertiaDep) -> InertiaResponse: @router.get( "/admin", response_model=None, - dependencies=[Depends(RequiresPermission("users.manage"))], + dependencies=[Depends(RequiresPermission(PERM_USERS_MANAGE))], ) async def admin_index( request: Request, @@ -116,7 +129,7 @@ async def admin_index( ) -> InertiaResponse: users, total = await service.list_users(page=page, per_page=per_page, search=q) return await inertia.render( - "Users/Users/Index", + _PAGE_ADMIN_INDEX, { "users": [u.model_dump(mode="json") for u in users], "pagination": {"page": page, "per_page": per_page, "total": total}, @@ -129,14 +142,14 @@ async def admin_index( @router.get( "/admin/invite", response_model=None, - dependencies=[Depends(RequiresPermission("users.manage"))], + dependencies=[Depends(RequiresPermission(PERM_USERS_MANAGE))], ) async def admin_invite_page( request: Request, inertia: InertiaDep, ) -> InertiaResponse: return await inertia.render( - "Users/Users/Invite", + _PAGE_ADMIN_INVITE, { "roles": [{"id": r.id, "name": r.name} for r in await get_roles_cache(request.app)], }, @@ -146,7 +159,7 @@ async def admin_invite_page( @router.get( "/admin/{user_id}", response_model=None, - dependencies=[Depends(RequiresPermission("users.manage"))], + dependencies=[Depends(RequiresPermission(PERM_USERS_MANAGE))], ) async def admin_edit_page( user_id: str, @@ -162,7 +175,7 @@ async def admin_edit_page( if user_item is None: raise HTTPException(status_code=404) return await inertia.render( - "Users/Users/Edit", + _PAGE_ADMIN_EDIT, { "user": user_item.model_dump(mode="json"), "roles": [{"id": r.id, "name": r.name} for r in await get_roles_cache(request.app)], diff --git a/modules/users/users/middleware.py b/modules/users/users/middleware.py index 5c87a92d..2f0d2ee4 100644 --- a/modules/users/users/middleware.py +++ b/modules/users/users/middleware.py @@ -27,11 +27,16 @@ from starlette.responses import RedirectResponse from starlette.types import ASGIApp, Receive, Scope, Send +from users.constants import SESSION_USER_ID_KEY from users.models import User logger = logging.getLogger(__name__) SESSION_USER_CTX_KEY = "user_ctx" +_SESSION_USER_ID_KEY = SESSION_USER_ID_KEY +_SESSION_NEXT_KEY = "next" +_SCOPE_HTTP = "http" +_LOGIN_REDIRECT = "/users/login" # Paths that don't require authentication. PUBLIC_PATHS = ( @@ -65,7 +70,7 @@ def __init__(self, app: ASGIApp) -> None: self.app = app async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": + if scope["type"] != _SCOPE_HTTP: await self.app(scope, receive, send) return @@ -73,7 +78,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: is_public = any(path.startswith(p) for p in PUBLIC_PATHS) or path in EXACT_PUBLIC_PATHS session = scope["session"] - raw_user_id = session.get("user_id") + raw_user_id = session.get(_SESSION_USER_ID_KEY) user_ctx: UserContext | None = None if raw_user_id: @@ -85,22 +90,22 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: user_uuid = uuid.UUID(user_id_str) except (ValueError, TypeError): logger.warning("Invalid user_id in session: %r", raw_user_id) - session.pop("user_id", None) + session.pop(_SESSION_USER_ID_KEY, None) session.pop(SESSION_USER_CTX_KEY, None) user_ctx = None else: user_ctx = await self._load_user(scope, user_uuid) if user_ctx is None: # User was deleted / disabled since session creation. - session.pop("user_id", None) + session.pop(_SESSION_USER_ID_KEY, None) session.pop(SESSION_USER_CTX_KEY, None) else: session[SESSION_USER_CTX_KEY] = user_ctx.to_session_dict() if user_ctx is None and not is_public: request = Request(scope) - session["next"] = str(request.url) - response = RedirectResponse("/users/login", status_code=302) + session[_SESSION_NEXT_KEY] = str(request.url) + response = RedirectResponse(_LOGIN_REDIRECT, status_code=302) await response(scope, receive, send) return diff --git a/modules/users/users/module.py b/modules/users/users/module.py index 033ee153..1e0c69a2 100644 --- a/modules/users/users/module.py +++ b/modules/users/users/module.py @@ -9,16 +9,35 @@ from simple_module_core.module import ModuleBase, ModuleMeta from simple_module_core.permissions import PermissionRegistry +from users.constants import ( + ADMIN_ROLE_NAME, + PERM_USERS_MANAGE, + PERM_USERS_SELF_PROFILE, + USER_ROLE_NAME, +) + if TYPE_CHECKING: from fastapi import FastAPI +_MODULE_DEPENDENCY_AUTH = "Auth" + +# Menu URLs +_URL_USERS_ADMIN = "/users/admin" +_URL_USERS_ME = "/users/me" +_URL_USERS_LOGOUT = "/users/logout" + +# Menu icons +_ICON_USERS = "users" +_ICON_USER = "user" +_ICON_LOG_OUT = "log-out" + class UsersModule(ModuleBase): meta = ModuleMeta( name="Users", route_prefix="/api/users", view_prefix="/users", - depends_on=["Auth"], + depends_on=[_MODULE_DEPENDENCY_AUTH], ) def register_settings(self, app: FastAPI) -> None: @@ -31,28 +50,28 @@ def register_settings(self, app: FastAPI) -> None: def register_permissions(self, registry: PermissionRegistry) -> None: registry.add_group( "Users", - ["users.manage", "users.self.profile"], + [PERM_USERS_MANAGE, PERM_USERS_SELF_PROFILE], ) - registry.map_role("user", ["users.self.profile"]) + registry.map_role(USER_ROLE_NAME, [PERM_USERS_SELF_PROFILE]) def register_menu_items(self, registry: MenuRegistry) -> None: # Admin-only user management registry.add( MenuItem( label="Users", - url="/users/admin", - icon="users", + url=_URL_USERS_ADMIN, + icon=_ICON_USERS, order=30, section=MenuSection.SIDEBAR, - roles=["admin"], + roles=[ADMIN_ROLE_NAME], ) ) # Self-service: profile + logout live in the user dropdown. registry.add( MenuItem( label="Profile", - url="/users/me", - icon="user", + url=_URL_USERS_ME, + icon=_ICON_USER, order=990, section=MenuSection.USER_DROPDOWN, ) @@ -60,8 +79,8 @@ def register_menu_items(self, registry: MenuRegistry) -> None: registry.add( MenuItem( label="Logout", - url="/users/logout", - icon="log-out", + url=_URL_USERS_LOGOUT, + icon=_ICON_LOG_OUT, order=999, section=MenuSection.USER_DROPDOWN, method="post", diff --git a/scripts/check_hardcoded_strings.py b/scripts/check_hardcoded_strings.py new file mode 100644 index 00000000..06773d2c --- /dev/null +++ b/scripts/check_hardcoded_strings.py @@ -0,0 +1,163 @@ +"""Enforce that magic strings (permissions, role names, Inertia pages, module +dependencies) are declared as named constants rather than inline literals. + +Violation patterns — checked only in non-test, non-constants .py files: + + - RequiresPermission("...") → use a PERM_* constant + - registry.map_role("...", ...) → use a ROLE_NAME constant + - registry.add_group(..., [...]) → permission items must be constants + - inertia.render("Module/Page") → use a _PAGE_* constant + - depends_on=[..., "Module", ...] → use a _MODULE_* constant + +Usage: + uv run python scripts/check_hardcoded_strings.py + uv run python scripts/check_hardcoded_strings.py --root some/subdir +""" + +from __future__ import annotations + +import argparse +import io +import re +import subprocess +import sys +import tokenize +from collections.abc import Iterable, Sequence +from pathlib import Path + +# Each rule is (pattern, message). The pattern is matched against each line; +# a match is a violation unless the file is excluded. +_RULES: list[tuple[re.Pattern[str], str]] = [ + ( + re.compile(r'RequiresPermission\s*\(\s*"[a-z_]+\.[a-z_]+"'), + "permission string literal in RequiresPermission() — use a PERM_* constant", + ), + ( + re.compile(r'registry\.map_role\s*\(\s*"[a-z_]+"'), + "role name literal in registry.map_role() — use a ROLE_NAME constant", + ), + ( + re.compile(r'registry\.add_group\s*\([^)]*"[a-z_]+\.[a-z_]+"'), + "permission string literal in registry.add_group() — use a PERM_* constant", + ), + ( + re.compile(r'inertia\.render\s*\(\s*"[A-Z][A-Za-z]+/[A-Za-z/]+"'), + "Inertia page identifier literal in render() — use a _PAGE_* constant", + ), + ( + re.compile(r'depends_on\s*=\s*\[[^\]]*"[A-Z][A-Za-z]+"'), + "module name literal in depends_on — use a _MODULE_* constant", + ), +] + +# Files whose path contains any of these substrings are skipped entirely. +_SKIP_PATH_PARTS = ("tests/", "test_", "/constants.py", "scripts/") + + +def _should_skip(path: Path, root: Path) -> bool: + rel = path.as_posix() + return any(part in rel for part in _SKIP_PATH_PARTS) + + +def _string_literal_lines(source: str) -> set[int]: + """Return line numbers that are part of a string/docstring token.""" + inside: set[int] = set() + try: + tokens = tokenize.generate_tokens(io.StringIO(source).readline) + for tok_type, _, (start_row, _), (end_row, _), _ in tokens: + if tok_type == tokenize.STRING: + inside.update(range(start_row, end_row + 1)) + except tokenize.TokenError: + pass + return inside + + +def _check_file(path: Path) -> list[tuple[int, str, str]]: + """Return a list of (line_number, line, message) violations.""" + violations: list[tuple[int, str, str]] = [] + try: + source = path.read_text(encoding="utf-8", errors="replace") + except OSError: + return violations + string_lines = _string_literal_lines(source) + for lineno, line in enumerate(source.splitlines(), start=1): + stripped = line.strip() + if stripped.startswith("#") or lineno in string_lines: + continue + for pattern, message in _RULES: + if pattern.search(line): + violations.append((lineno, line.rstrip(), message)) + break + return violations + + +def _list_git_tracked_files(root: Path) -> list[Path]: + result = subprocess.run( + ["git", "ls-files"], + cwd=root, + capture_output=True, + text=True, + check=True, + ) + return [root / line for line in result.stdout.splitlines() if line.endswith(".py")] + + +def _walk_filesystem(root: Path) -> list[Path]: + return list(root.rglob("*.py")) + + +def _collect_candidates(root: Path, use_git: bool) -> list[Path]: + return _list_git_tracked_files(root) if use_git else _walk_filesystem(root) + + +def find_violations( + paths: Iterable[Path], + root: Path, +) -> list[tuple[Path, int, str, str]]: + """Return (path, lineno, line, message) for every violation.""" + results: list[tuple[Path, int, str, str]] = [] + for path in paths: + if _should_skip(path, root): + continue + for lineno, line, message in _check_file(path): + results.append((path, lineno, line, message)) + return results + + +def _parse_args(argv: Sequence[str] | None) -> argparse.Namespace: + description = (__doc__ or "").splitlines()[0] + parser = argparse.ArgumentParser(description=description) + parser.add_argument( + "--root", + type=Path, + default=Path.cwd(), + help="repo root to scan (default: cwd)", + ) + parser.add_argument( + "--no-git", + action="store_true", + help="walk the filesystem instead of using git ls-files", + ) + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + args = _parse_args(argv) + root: Path = args.root + candidates = _collect_candidates(root, use_git=not args.no_git) + violations = find_violations(candidates, root) + if not violations: + print("OK: no hardcoded magic strings found.") + return 0 + + print(f"FAIL: {len(violations)} hardcoded magic string(s) found:\n") + for path, lineno, line, message in violations: + rel = path.relative_to(root) if path.is_absolute() else path + print(f" {rel.as_posix()}:{lineno}: {message}") + print(f" {line}") + print() + return 1 + + +if __name__ == "__main__": + sys.exit(main())