feat(auth): implement comprehensive JWT and Basic authentication

README.md

@@ -304,6 +304,22 @@ class AdminOnlyEndpoint(RestEndpoint):
 2. Permission class `.has_permission(request)` — checks `request.state.user`
 3. Returns `401` if authentication fails, `403` if permission denied
 
+**Login and token endpoints:** When using `JWTAuthentication` or `BasicAuthentication`, pass `login_validator` to obtain automatic `/auth/login` and `/auth/token` endpoints:
+
+```python
+def my_validator(username: str, password: str):
+    # Return user payload dict or None
+    user = db.query(User).filter_by(username=username).first()
+    if user and user.check_password(password):
+        return {"sub": str(user.id), "is_admin": user.is_admin}
+    return None
+
+app = LightApi(engine=engine, login_validator=my_validator)
+app.register({"/secrets": ProtectedEndpoint})
+# POST /auth/login and POST /auth/token now accept {"username":"...","password":"..."}
+# JWT mode: 200 {"token":"...","user":{...}}; Basic-only: 200 {"user":{...}}
+```
+
 **Built-in permission classes:**
 
 | Class | Condition |

docs/examples/yaml-configuration.md

@@ -22,13 +22,20 @@ defaults:
   authentication:
     backend: JWTAuthentication
     permission: IsAuthenticated
+    jwt_expiration: 3600
+    jwt_extra_claims: [sub, email]
+    jwt_algorithm: HS256
   pagination:
     style: page_number
     page_size: 20
 
 middleware:
   - CORSMiddleware
 
+auth:
+  auth_path: /auth
+  login_validator: myapp.validators.validate_login
+
 endpoints:
   - route: /products
     fields:
@@ -130,8 +137,13 @@ python -c "from lightapi import LightApi; LightApi.from_config('lightapi.yaml').
 |-------|------|-------------|
 | `database.url` | string | SQLAlchemy URL. Supports `${VAR}` substitution. |
 | `cors_origins` | list | CORS allowed origins. |
-| `defaults.authentication.backend` | string | Auth backend class name. |
+| `defaults.authentication.backend` | string | Auth backend class name (`JWTAuthentication`, `BasicAuthentication`). |
 | `defaults.authentication.permission` | string | Permission class name. |
+| `defaults.authentication.jwt_expiration` | int | JWT token expiration in seconds (JWT only). |
+| `defaults.authentication.jwt_extra_claims` | list | Claims to include in token payload (JWT only). |
+| `defaults.authentication.jwt_algorithm` | string | JWT algorithm (HS256, HS384, HS512, RS256, RS384, RS512, ES256, ES384, ES512). |
+| `auth.auth_path` | string | Path prefix for `/login` and `/token` (default `/auth`). |
+| `auth.login_validator` | string | Dotted path to credential validator callable (e.g. `myapp.validators.check_user`). |
 | `defaults.pagination.style` | string | `page_number` or `cursor`. |
 | `defaults.pagination.page_size` | int | Rows per page. |
 | `middleware` | list | Class names resolved at startup. |

lightapi/__init__.py

@@ -1,6 +1,12 @@
 """LightAPI v2 public API."""
 
-from lightapi.auth import AllowAny, IsAdminUser, IsAuthenticated, JWTAuthentication
+from lightapi.auth import (
+    AllowAny,
+    BasicAuthentication,
+    IsAdminUser,
+    IsAuthenticated,
+    JWTAuthentication,
+)
 from lightapi.cache import RedisCache
 from lightapi.config import Authentication, Cache, Filtering, Pagination, Serializer
 
@@ -33,6 +39,7 @@
     "Pagination",
     "Serializer",
     # Auth
+    "BasicAuthentication",
     "JWTAuthentication",
     "AllowAny",
     "IsAuthenticated",

lightapi/_login.py

@@ -0,0 +1,135 @@
+"""Login and token endpoint handlers."""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+from collections.abc import Callable
+from typing import Any, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+# JWTAuthentication imported locally where needed to avoid circular import
+
+logger = logging.getLogger(__name__)
+
+
+class LoginRequest(BaseModel):
+    """Request body for POST /auth/login and /auth/token."""
+
+    model_config = ConfigDict(frozen=True)
+
+    username: str = Field(min_length=1)
+    password: str = Field(min_length=1)
+
+
+def _parse_basic_header(auth_header: str) -> tuple[str, str] | None:
+    """
+    Decode Authorization: Basic header.
+
+    Returns (username, password) or None if malformed.
+    """
+    if not auth_header.lower().startswith("basic "):
+        return None
+    try:
+        token = auth_header.split(" ", 1)[1]
+        decoded = base64.b64decode(token).decode("utf-8")
+    except (ValueError, IndexError, UnicodeDecodeError):
+        return None
+    parts = decoded.split(":", 1)
+    if len(parts) != 2:
+        return None
+    return parts[0], parts[1]
+
+
+async def _parse_credentials(request: Request) -> tuple[str, str] | None:
+    """
+    Extract (username, password) from request.
+
+    - If Authorization: Basic present: returns (u, p) or None if malformed.
+    - If no Basic header: reads body, validates with LoginRequest.
+      Returns (u, p) if valid. Raises ValidationError for body (caller returns 422).
+    - None means malformed Basic (caller returns 401).
+    """
+    auth_header = request.headers.get("Authorization")
+    if auth_header and auth_header.startswith("Basic "):
+        return _parse_basic_header(auth_header)
+
+    body = await _read_body(request)
+    parsed = LoginRequest.model_validate(body if body else {})
+    return parsed.username, parsed.password
+
+
+async def _read_body(request: Request) -> dict[str, Any]:
+    """Read JSON body; return {} on empty or invalid."""
+    try:
+        body = await request.body()
+        return json.loads(body) if body else {}
+    except (json.JSONDecodeError, TypeError):
+        return {}
+
+
+async def login_handler(
+    request: Request,
+    *,
+    login_validator: Callable[[str, str], dict[str, Any] | None],
+    has_jwt: bool,
+    jwt_expiration: int | None = None,
+    jwt_extra_claims: list[str] | None = None,
+    jwt_algorithm: str | None = None,
+    rate_limiter: Optional[Any] = None,
+) -> JSONResponse:
+    """
+    Handle POST /auth/login and POST /auth/token.
+
+    Returns 422 for body validation, 401 for malformed Basic or invalid credentials,
+    500 for validator exception, 200 with token+user (JWT) or user only (Basic).
+    """
+    # Apply rate limiting if a rate limiter is provided
+    if rate_limiter is not None:
+        is_limited, window = rate_limiter.is_rate_limited(request, endpoint="auth")
+        if is_limited:
+            return rate_limiter.get_rate_limit_response(request, window)
+
+    if request.method != "POST":
+        return JSONResponse(
+            {"detail": "method not allowed"},
+            status_code=405,
+            headers={"Allow": "POST"},
+        )
+
+    try:
+        creds = await _parse_credentials(request)
+    except ValidationError as exc:
+        return JSONResponse({"detail": exc.errors()}, status_code=422)
+
+    if creds is None:
+        return JSONResponse({"detail": "Invalid credentials"}, status_code=401)
+
+    username, password = creds
+    try:
+        payload = login_validator(username, password)
+    except Exception as e:
+        logger.exception("login_validator raised: %s", e)
+        raise
+
+    if payload is None:
+        return JSONResponse({"detail": "Invalid credentials"}, status_code=401)
+
+    if has_jwt:
+        from lightapi.auth import JWTAuthentication
+
+        jwt_auth = JWTAuthentication(algorithm=jwt_algorithm)
+        if jwt_extra_claims and isinstance(payload, dict):
+            token_payload = {k: payload[k] for k in jwt_extra_claims if k in payload}
+            if not token_payload:
+                token_payload = payload
+        else:
+            token_payload = payload
+        token = jwt_auth.generate_token(token_payload, expiration=jwt_expiration)
+        return JSONResponse({"token": token, "user": payload})
+
+    return JSONResponse({"user": payload})

lightapi/_registry.py

@@ -7,31 +7,53 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
+from typing import Any, cast
+
 from sqlalchemy import MetaData
 from sqlalchemy.orm import registry
 
-_registry: registry | None = None
-_metadata: MetaData | None = None
-_engine: object | None = None
+LoginValidator = Callable[[str, str], dict[str, Any] | None]
+
+_state: dict[str, object | None] = {
+    "registry": None,
+    "metadata": None,
+    "engine": None,
+    "login_validator": None,
+}
 
 
 def get_registry_and_metadata() -> tuple[registry, MetaData]:
-    global _registry, _metadata
-    if _registry is None or _metadata is None:
-        _metadata = MetaData()
-        _registry = registry(metadata=_metadata)
-    return _registry, _metadata
+    reg = cast(registry | None, _state["registry"])
+    meta = cast(MetaData | None, _state["metadata"])
+    if reg is None or meta is None:
+        meta = MetaData()
+        reg = registry(metadata=meta)
+        _state["metadata"] = meta
+        _state["registry"] = reg
+    return reg, meta
 
 
 def set_engine(engine: object) -> None:
-    global _engine
-    _engine = engine
+    _state["engine"] = engine
 
 
 def get_engine() -> object:
-    if _engine is None:
+    engine = _state["engine"]
+    if engine is None:
         raise RuntimeError(
             "No engine configured. Call LightApi(engine=...) or ensure "
             "database connection is set before the first request."
         )
-    return _engine
+    return engine
+
+
+def set_login_validator(validator: LoginValidator) -> None:
+    _state["login_validator"] = validator
+
+
+def get_login_validator() -> LoginValidator | None:
+    validator = _state["login_validator"]
+    if validator is None:
+        return None
+    return cast(LoginValidator, validator)