diff --git a/pygitweb/auth.schema.json b/pygitweb/auth.schema.json index 500e981..18d877c 100644 --- a/pygitweb/auth.schema.json +++ b/pygitweb/auth.schema.json @@ -98,6 +98,40 @@ "assume_anonymous": { "type": "boolean", "description": "If true, unauthenticated requests assume this user for read permissions (e.g. public viewer role)." + }, + "keys": { + "type": "array", + "description": "Smart HTTP credentials for this user. Git sends the local username or e-mail as the Basic username and the key as the password. Login passwords are not accepted.", + "items": { + "type": "object", + "required": ["id", "secret", "time_issued"], + "properties": { + "id": { + "type": "string", + "minLength": 1, + "description": "Label unique among this user's keys." + }, + "secret": { + "type": "string", + "minLength": 1, + "description": "scrypt hash of the key, same format as pass_hash. Plaintext is not accepted." + }, + "readonly": { + "type": "boolean", + "description": "If true, the key can fetch (git-upload-pack) but not push (git-receive-pack)." + }, + "time_issued": { + "type": "string", + "format": "date-time", + "description": "When the key was issued (ISO-8601)." + }, + "time_expiry": { + "type": "string", + "format": "date-time", + "description": "When the key stops authenticating (ISO-8601). Omit for a key that does not expire. A key is expired at this instant." + } + } + } } } } diff --git a/pygitweb/auth_keys_test.py b/pygitweb/auth_keys_test.py new file mode 100644 index 0000000..05ef092 --- /dev/null +++ b/pygitweb/auth_keys_test.py @@ -0,0 +1,194 @@ +import json +from datetime import UTC, datetime, timedelta +from pathlib import Path +from unittest.mock import patch + +import pytest +from fastapi.testclient import TestClient +from pydantic import ValidationError +from pygitweb.auth import authenticate_http_key +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.config import Settings, settings +from pygitweb.conftest import set_client_session +from pygitweb.main import app +from pygitweb.password_hash import is_password_hash, verify_password +from pygitweb.sessions import create_session + +_SALT = "0123456789abcdef" + + +def _config() -> AuthConfig: + return AuthConfig( + auth_mode="local", + salt=_SALT, + local_users=[LocalUser(user="admin", password="secret", email="admin@example.com")], + ) + + +def test_api_key_settings_default_to_two_keys_without_expiry() -> None: + cfg = Settings(AUTH=False) + assert cfg.API_KEY_MAX_COUNT == 2 + assert cfg.API_KEY_EXPIRY_DAYS == 0 + with pytest.raises(ValidationError): + Settings(AUTH=False, API_KEY_MAX_COUNT=-1) + + +def test_issue_api_key_returns_plaintext_and_stores_hash(tmp_path: Path) -> None: + cfg = _config() + path = tmp_path / "auth.json" + client = TestClient(app) + with ( + patch.object(settings, "AUTH", True), + patch.object(settings, "AUTH_CONFIG", str(path)), + patch("pygitweb.auth_config.auth_config", cfg), + patch("pygitweb.auth.auth_config", cfg), + patch("pygitweb.main.auth_config", cfg), + ): + set_client_session(client, "admin") + page = client.get("/user") + assert page.status_code == 200 + assert "API keys" in page.text + assert "Create key" in page.text + issued = client.post("/user/keys", json={"id": "laptop", "readonly": True}) + assert issued.status_code == 200, issued.text + body = issued.json() + assert body["id"] == "laptop" + assert body["readonly"] is True + plaintext = body["secret"] + assert isinstance(plaintext, str) + assert not plaintext.startswith("scrypt$") + assert body["time_expiry"] is None + stored = cfg.local_users[0].keys[0] + assert stored.secret != plaintext + assert is_password_hash(stored.secret) + assert verify_password(plaintext, stored.secret, config_salt=_SALT) + on_disk = json.loads(path.read_text(encoding="utf-8")) + assert on_disk["local_users"][0]["keys"][0]["secret"] == stored.secret + assert plaintext not in path.read_text(encoding="utf-8") + matched = authenticate_http_key("admin", plaintext) + assert matched is not None + assert matched[1] is True + again = client.get("/user") + assert "laptop" in again.text + assert plaintext not in again.text + + +def test_issue_api_key_enforces_limit_expiry_and_duplicate_id(tmp_path: Path) -> None: + cfg = _config() + path = tmp_path / "auth.json" + client = TestClient(app) + with ( + patch.object(settings, "AUTH", True), + patch.object(settings, "AUTH_CONFIG", str(path)), + patch.object(settings, "API_KEY_MAX_COUNT", 2), + patch.object(settings, "API_KEY_EXPIRY_DAYS", 7), + patch("pygitweb.auth_config.auth_config", cfg), + patch("pygitweb.auth.auth_config", cfg), + patch("pygitweb.main.auth_config", cfg), + ): + set_client_session(client, "admin") + before = datetime.now(UTC) + first = client.post("/user/keys", json={"id": "alpha-key"}) + assert first.status_code == 200, first.text + duplicate = client.post("/user/keys", json={"id": "alpha-key"}) + second = client.post("/user/keys", json={"id": "beta-key"}) + assert second.status_code == 200, second.text + third = client.post("/user/keys", json={"id": "gamma-key"}) + assert duplicate.status_code == 409 + assert third.status_code == 409 + assert len(cfg.local_users[0].keys) == 2 + expiry = datetime.fromisoformat(first.json()["time_expiry"]) + assert before + timedelta(days=7) <= expiry <= datetime.now(UTC) + timedelta(days=7) + assert second.json()["time_expiry"] is not None + revoked = client.delete("/user/keys", params={"id": "alpha-key"}) + assert revoked.status_code == 204 + assert [key.id for key in cfg.local_users[0].keys] == ["beta-key"] + page = client.get("/user") + assert "alpha-key" not in page.text + assert "beta-key" in page.text + replaced = client.post("/user/keys", json={"id": "alpha-key"}) + assert replaced.status_code == 200 + assert replaced.json()["id"] == "alpha-key" + + +def test_issue_api_key_requires_a_local_account_and_a_session(tmp_path: Path) -> None: + cfg = _config() + path = tmp_path / "auth.json" + client = TestClient(app) + with ( + patch.object(settings, "AUTH", True), + patch.object(settings, "AUTH_CONFIG", str(path)), + patch("pygitweb.auth_config.auth_config", cfg), + patch("pygitweb.auth.auth_config", cfg), + patch("pygitweb.main.auth_config", cfg), + ): + missing = client.post("/user/keys", json={"id": "laptop"}) + assert missing.status_code == 401 + token = create_session("nobody", auth_method="local") + unknown = client.post( + "/user/keys", + json={"id": "laptop"}, + headers={"Authorization": f"Bearer {token}"}, + ) + assert unknown.status_code == 404 + page = client.get("/user", headers={"Authorization": f"Bearer {token}"}) + assert "cannot be stored" in page.text + email_token = create_session("other-name", email="Admin@Example.com", auth_method="oauth") + by_email = client.post( + "/user/keys", + json={"id": "mail-key"}, + headers={"Authorization": f"Bearer {email_token}"}, + ) + assert by_email.status_code == 200, by_email.text + assert cfg.local_users[0].keys[0].id == "mail-key" + missing_key = client.delete( + "/user/keys", params={"id": "missing"}, headers={"Authorization": f"Bearer {email_token}"} + ) + assert missing_key.status_code == 404 + revoke_unknown = client.delete( + "/user/keys", params={"id": "mail-key"}, headers={"Authorization": f"Bearer {token}"} + ) + assert revoke_unknown.status_code == 404 + with patch.object(settings, "AUTH", False): + disabled = client.post("/user/keys", json={"id": "laptop"}) + assert disabled.status_code == 400 + disabled_revoke = client.delete("/user/keys", params={"id": "laptop"}) + assert disabled_revoke.status_code == 400 + + +def test_issue_api_key_keeps_memory_unchanged_when_save_fails(tmp_path: Path) -> None: + cfg = _config() + path = tmp_path / "auth.json" + client = TestClient(app, raise_server_exceptions=False) + with ( + patch.object(settings, "AUTH", True), + patch.object(settings, "AUTH_CONFIG", str(path)), + patch("pygitweb.auth_config.auth_config", cfg), + patch("pygitweb.auth.auth_config", cfg), + patch("pygitweb.main.auth_config", cfg), + patch("pygitweb.auth.write_auth_config", side_effect=OSError("disk")), + ): + set_client_session(client, "admin") + response = client.post("/user/keys", json={"id": "laptop"}) + assert response.status_code == 500 + assert cfg.local_users[0].keys == [] + + +def test_revoke_api_key_keeps_memory_unchanged_when_save_fails(tmp_path: Path) -> None: + cfg = _config() + path = tmp_path / "auth.json" + client = TestClient(app, raise_server_exceptions=False) + with ( + patch.object(settings, "AUTH", True), + patch.object(settings, "AUTH_CONFIG", str(path)), + patch("pygitweb.auth_config.auth_config", cfg), + patch("pygitweb.auth.auth_config", cfg), + patch("pygitweb.main.auth_config", cfg), + ): + set_client_session(client, "admin") + issued = client.post("/user/keys", json={"id": "laptop"}) + assert issued.status_code == 200, issued.text + with patch("pygitweb.auth.write_auth_config", side_effect=OSError("disk")): + response = client.delete("/user/keys", params={"id": "laptop"}) + assert response.status_code == 500 + assert [key.id for key in cfg.local_users[0].keys] == ["laptop"] diff --git a/pygitweb/settings.schema.json b/pygitweb/settings.schema.json index 9eb98d6..a3b7735 100644 --- a/pygitweb/settings.schema.json +++ b/pygitweb/settings.schema.json @@ -61,6 +61,16 @@ "type": "string" }, "description": "Distribution names of installed plugins whose entry points are active. Empty means all discovered plugins are off." + }, + "API_KEY_MAX_COUNT": { + "type": "integer", + "minimum": 0, + "description": "Maximum number of Smart HTTP API keys each local user may hold. Default 2." + }, + "API_KEY_EXPIRY_DAYS": { + "type": "integer", + "minimum": 0, + "description": "Lifetime of a newly issued API key, in days. 0 means the key does not expire." } }, "required": [ @@ -76,6 +86,8 @@ "AUTH", "AUTH_CONFIG", "SETTINGS_CONFIG", - "ENABLED_PLUGINS" + "ENABLED_PLUGINS", + "API_KEY_MAX_COUNT", + "API_KEY_EXPIRY_DAYS" ] } diff --git a/pygitweb/src/pygitweb/auth.py b/pygitweb/src/pygitweb/auth.py index 1ad01ce..d8ab689 100644 --- a/pygitweb/src/pygitweb/auth.py +++ b/pygitweb/src/pygitweb/auth.py @@ -14,24 +14,34 @@ FastAPI security integration from __future__ import annotations +import base64 +import binascii import hmac +import secrets +import threading from collections.abc import Callable -from typing import Annotated +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Annotated, Literal from fastapi import APIRouter, Depends, Form, HTTPException, Query, status -from fastapi.responses import HTMLResponse, RedirectResponse +from fastapi.responses import HTMLResponse, RedirectResponse, Response from fastapi.security import APIKeyCookie, OAuth2PasswordBearer, OAuth2PasswordRequestForm -from pydantic import BaseModel +from pydantic import BaseModel, Field, field_validator from pygitweb.auth_config import ( + HttpKey, + LocalUser, + auth_config_path, is_auth_configured, is_local_auth_available, is_oauth_auth_available, + write_auth_config, ) from pygitweb.auth_oauth import OAUTH_STATE_COOKIE_NAME, oauth_router from pygitweb.config import auth_config, settings from pygitweb.gravatar import gravatar_url -from pygitweb.password_hash import verify_password +from pygitweb.password_hash import hash_password, verify_password from pygitweb.permissions import Permission, PermissionPrincipal, has_branch_read_permission from pygitweb.projects import git_get_project_owner from pygitweb.sessions import create_session, get_session, revoke_session @@ -56,6 +66,30 @@ class User(BaseModel): email: str | None = None +SmartHttpAuthError = Literal["unauthorized", "unavailable"] +_INVALID_KEY_ACTOR = "INVALID_KEY" + + +@dataclass(frozen=True, slots=True) +class SmartHttpCredentials: + required: bool + principal: PermissionPrincipal | None = None + readonly: bool = False + actor: str = "" + error: SmartHttpAuthError | None = None + + +@dataclass(frozen=True, slots=True) +class PresentedHttpKey: + principal: PermissionPrincipal + readonly: bool + tail: str + expired: bool + + +_SMART_HTTP_CHALLENGE = {"WWW-Authenticate": 'Basic realm="pygitweb", Bearer'} + + def principal_from_session(token: str | None) -> PermissionPrincipal | None: if not token: return None @@ -96,11 +130,8 @@ def has_permission( return auth_config.oauth_permissions.has_permission(identity, permission, project) -def _owner_matches(project: str, token: str | None) -> bool: - if not settings.AUTH or not token: - return False - principal = principal_from_session(token) - if principal is None: +def _owner_matches_principal(project: str, principal: PermissionPrincipal) -> bool: + if not settings.AUTH: return False owner = git_get_project_owner(project) if not owner or not owner.strip(): @@ -112,6 +143,31 @@ def _owner_matches(project: str, token: str | None) -> bool: return owner_norm.lower() == identity.lower() +def _owner_matches(project: str, token: str | None) -> bool: + if not token: + return False + principal = principal_from_session(token) + if principal is None: + return False + return _owner_matches_principal(project, principal) + + +def principal_can_read_project(project: str, principal: PermissionPrincipal) -> bool: + if not settings.AUTH: + return True + if _owner_matches_principal(project, principal): + return True + return auth_config.oauth_permissions.has_permission(principal.identity, Permission.READ, project) + + +def principal_can_write_project(project: str, principal: PermissionPrincipal) -> bool: + if not settings.AUTH: + return True + if _owner_matches_principal(project, principal): + return True + return auth_config.oauth_permissions.has_permission(principal.identity, Permission.WRITE, project) + + def can_read_project(project: str, token: str | None) -> bool: if not settings.AUTH: return True @@ -253,6 +309,163 @@ def authenticate_local_user(username: str, password: str) -> User | None: return None +def _const_eq(left: str, right: str) -> bool: + if len(left) != len(right): + return False + return hmac.compare_digest(left, right) + + +def _http_key_username_matches(user: str, email: str | None, username: str) -> bool: + if _const_eq(user, username): + return True + if not email: + return False + return _const_eq(email.casefold(), username.casefold()) + + +def _key_owner_name(principal: PermissionPrincipal) -> str: + email = principal.email.strip() if principal.email else "" + return email or principal.username + + +def _http_key_actor(matched: PresentedHttpKey) -> str: + owner = _key_owner_name(matched.principal) + if matched.expired: + return f"{owner} {matched.tail} (EXPIRED)" + return f"{owner} {matched.tail}" + + +def match_presented_http_key(username: str, password: str) -> PresentedHttpKey | None: + presented_user = username.strip() + presented_secret = password.strip() + if not presented_user or not presented_secret: + return None + tail = presented_secret[-4:] + now = datetime.now(UTC) + expired_match: PresentedHttpKey | None = None + for entry in auth_config.local_users: + if not _http_key_username_matches(entry.user, entry.email, presented_user): + continue + principal = PermissionPrincipal(username=entry.user, email=entry.email) + for key in entry.keys: + if not verify_password(presented_secret, key.secret, config_salt=auth_config.salt): + continue + if key.is_expired(now): + if expired_match is None: + expired_match = PresentedHttpKey( + principal=principal, + readonly=key.readonly, + tail=tail, + expired=True, + ) + continue + return PresentedHttpKey(principal=principal, readonly=key.readonly, tail=tail, expired=False) + return expired_match + + +def authenticate_http_key(username: str, password: str) -> tuple[PermissionPrincipal, bool] | None: + matched = match_presented_http_key(username, password) + if matched is None or matched.expired: + return None + return matched.principal, matched.readonly + + +def _parse_basic_authorization(value: str) -> tuple[str, str] | None: + scheme, _, param = value.strip().partition(" ") + if scheme.lower() != "basic" or not param.strip(): + return None + try: + decoded = base64.b64decode(param.strip(), validate=True) + text = decoded.decode("utf-8") + except (binascii.Error, UnicodeDecodeError, ValueError): + return None + username, sep, password = text.partition(":") + if not sep: + return None + return username, password + + +def unauthorized_smart_http() -> HTTPException: + return HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Not authenticated", + headers=_SMART_HTTP_CHALLENGE, + ) + + +def raise_for_smart_http_credentials(creds: SmartHttpCredentials) -> None: + if creds.error == "unavailable": + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Authentication enabled but auth.json is not configured", + ) + if creds.error == "unauthorized": + raise unauthorized_smart_http() + + +def _assumable_public_username(project: str, *, write: bool) -> str: + if write: + return "" + anonymous = _anonymous_principal_for_permissions() + if anonymous is None or not can_read_project(project, None): + return "" + return anonymous.username + + +def load_smart_http_credentials( + authorization: str | None, + cookie_token: str | None, + *, + project: str, + write: bool, +) -> SmartHttpCredentials: + if not settings.AUTH: + return SmartHttpCredentials(required=False) + if not is_auth_configured(auth_config) and not any(user.keys for user in auth_config.local_users): + return SmartHttpCredentials(required=True, error="unavailable") + if authorization: + scheme, _, rest = authorization.strip().partition(" ") + scheme_name = scheme.lower() + if scheme_name == "basic": + parsed = _parse_basic_authorization(authorization) + if parsed is None: + return SmartHttpCredentials(required=True, actor=_INVALID_KEY_ACTOR, error="unauthorized") + matched = match_presented_http_key(parsed[0], parsed[1]) + if matched is None: + return SmartHttpCredentials(required=True, actor=_INVALID_KEY_ACTOR, error="unauthorized") + actor = _http_key_actor(matched) + if matched.expired: + return SmartHttpCredentials(required=True, actor=actor, error="unauthorized") + return SmartHttpCredentials( + required=True, + principal=matched.principal, + readonly=matched.readonly, + actor=actor, + ) + if scheme_name == "bearer": + bearer_principal = principal_from_session(rest.strip()) + if bearer_principal is None: + return SmartHttpCredentials(required=True, error="unauthorized") + return SmartHttpCredentials( + required=True, + principal=bearer_principal, + readonly=False, + actor=_key_owner_name(bearer_principal), + ) + return SmartHttpCredentials(required=True, error="unauthorized") + if cookie_token and cookie_token.strip(): + cookie_principal = principal_from_session(cookie_token) + if cookie_principal is None: + return SmartHttpCredentials(required=True, error="unauthorized") + return SmartHttpCredentials( + required=True, + principal=cookie_principal, + readonly=False, + actor=_key_owner_name(cookie_principal), + ) + return SmartHttpCredentials(required=True, actor=_assumable_public_username(project, write=write)) + + def decode_access_token(token: str) -> User | None: if not settings.AUTH: return None @@ -431,6 +644,134 @@ async def auth_status( return {"auth_enabled": True, "username": rec.username, "gravatar_url": avatar} +_API_KEY_LOCK = threading.Lock() + + +class IssueApiKeyRequest(BaseModel): + id: str = Field(min_length=1) + readonly: bool = False + + @field_validator("id", mode="before") + @classmethod + def _strip_id(cls, value: object) -> object: + if not isinstance(value, str): + return value + stripped = value.strip() + if not stripped: + raise ValueError("must be a non-empty string") + return stripped + + +class IssuedApiKey(BaseModel): + id: str + secret: str + readonly: bool + time_issued: datetime + time_expiry: datetime | None = None + + +def _local_user_for_principal(username: str, email: str | None) -> LocalUser | None: + for entry in auth_config.local_users: + if _const_eq(entry.user, username): + return entry + if not email: + return None + folded = email.casefold() + for entry in auth_config.local_users: + if entry.email and _const_eq(entry.email.casefold(), folded): + return entry + return None + + +def _key_expiry(issued: datetime) -> datetime | None: + days = settings.API_KEY_EXPIRY_DAYS + if days <= 0: + return None + return issued + timedelta(days=days) + + +def _persist_auth_config() -> None: + write_auth_config(auth_config_path(settings), auth_config) + + +def _public_key_rows(user: LocalUser) -> list[dict[str, str | bool | None]]: + rows: list[dict[str, str | bool | None]] = [] + for key in user.keys: + rows.append({ + "id": key.id, + "readonly": key.readonly, + "time_issued": key.time_issued.isoformat(), + "time_expiry": key.time_expiry.isoformat() if key.time_expiry else None, + }) + return rows + + +def issue_api_key(username: str, email: str | None, *, key_id: str, readonly: bool) -> IssuedApiKey: + with _API_KEY_LOCK: + user = _local_user_for_principal(username, email) + if user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys") + if len(user.keys) >= settings.API_KEY_MAX_COUNT: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"API key limit reached ({settings.API_KEY_MAX_COUNT})", + ) + if any(_const_eq(existing.id, key_id) for existing in user.keys): + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="An API key with this id already exists") + issued = datetime.now(UTC) + plaintext = secrets.token_urlsafe(32) + stored = HttpKey( + id=key_id, + secret=hash_password(plaintext, config_salt=auth_config.salt), + readonly=readonly, + time_issued=issued, + time_expiry=_key_expiry(issued), + ) + previous = list(user.keys) + user.keys = [*previous, stored] + try: + _persist_auth_config() + except Exception: + user.keys = previous + raise + return IssuedApiKey( + id=stored.id, + secret=plaintext, + readonly=stored.readonly, + time_issued=stored.time_issued, + time_expiry=stored.time_expiry, + ) + + +def revoke_api_key(username: str, email: str | None, key_id: str) -> None: + with _API_KEY_LOCK: + user = _local_user_for_principal(username, email) + if user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys") + remaining = [key for key in user.keys if not _const_eq(key.id, key_id)] + if len(remaining) == len(user.keys): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such API key") + previous = list(user.keys) + user.keys = remaining + try: + _persist_auth_config() + except Exception: + user.keys = previous + raise + + +def _session_identity(token: str | None) -> tuple[str, str | None]: + ensure_active_user_if_auth_enabled(token) + rec = get_session(token or "") + if rec is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Not authenticated", + headers={"WWW-Authenticate": "Bearer"}, + ) + return rec.username, rec.email + + @auth_router.get( "/user", response_class=HTMLResponse, @@ -457,13 +798,57 @@ async def user_account_page( u = decode_access_token(token or "") if token else None if u is None: return RedirectResponse(url="/login?next=/user", status_code=303) + rec = get_session(token or "") + email = rec.email if rec is not None else None + local_user = _local_user_for_principal(u.username, email) return render_page( "user_profile.html", title=f"{settings.SITE_NAME} - Account", username=u.username, + api_keys=_public_key_rows(local_user) if local_user is not None else None, + api_key_max=settings.API_KEY_MAX_COUNT, + api_key_expiry_days=settings.API_KEY_EXPIRY_DAYS, ) +@auth_router.post( + "/user/keys", + response_model=IssuedApiKey, + tags=[TAG_AUTH], + summary="Issue an API key", + description=( + "Creates a Smart HTTP API key for the signed-in local user. " + "The plaintext secret is returned once. Only its scrypt hash is stored." + ), +) +async def issue_user_api_key( + body: IssueApiKeyRequest, + token: Annotated[str | None, Depends(access_token_from_request)], +) -> IssuedApiKey: + if not settings.AUTH: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled") + username, email = _session_identity(token) + return issue_api_key(username, email, key_id=body.id, readonly=body.readonly) + + +@auth_router.delete( + "/user/keys", + status_code=status.HTTP_204_NO_CONTENT, + tags=[TAG_AUTH], + summary="Revoke an API key", + description="Removes one Smart HTTP API key from the signed-in local user.", +) +async def revoke_user_api_key( + token: Annotated[str | None, Depends(access_token_from_request)], + id: Annotated[str, Query(min_length=1)], +) -> Response: + if not settings.AUTH: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled") + username, email = _session_identity(token) + revoke_api_key(username, email, id.strip()) + return Response(status_code=status.HTTP_204_NO_CONTENT) + + @auth_router.get( "/users/me", response_model=User, diff --git a/pygitweb/src/pygitweb/auth_config.py b/pygitweb/src/pygitweb/auth_config.py index 97398aa..a6c3253 100644 --- a/pygitweb/src/pygitweb/auth_config.py +++ b/pygitweb/src/pygitweb/auth_config.py @@ -11,12 +11,13 @@ import os import secrets import stat import sys +from datetime import UTC, datetime from pathlib import Path -from typing import TYPE_CHECKING, Literal +from typing import TYPE_CHECKING, Literal, Self -from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator +from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator, model_validator -from pygitweb.password_hash import hash_password +from pygitweb.password_hash import hash_password, is_password_hash from pygitweb.permissions import PermissionsMap if TYPE_CHECKING: @@ -65,6 +66,48 @@ def generate_auth_salt() -> str: return secrets.token_hex(_AUTH_SALT_HEX_LEN // 2) +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +class HttpKey(BaseModel): + id: str = Field(min_length=1) + secret: str = Field(min_length=1) + readonly: bool = False + time_issued: datetime + time_expiry: datetime | None = None + + @field_validator("id", "secret", mode="before") + @classmethod + def _strip_required_text(cls, value: object) -> object: + if not isinstance(value, str): + return value + stripped = value.strip() + if not stripped: + raise ValueError("must be a non-empty string") + return stripped + + @field_validator("secret") + @classmethod + def _require_secret_hash(cls, value: str) -> str: + if not is_password_hash(value): + raise ValueError("secret must be a scrypt password hash") + return value + + @model_validator(mode="after") + def _expiry_not_before_issue(self) -> Self: + if self.time_expiry is not None and _as_utc(self.time_expiry) < _as_utc(self.time_issued): + raise ValueError("time_expiry is before time_issued") + return self + + def is_expired(self, now: datetime) -> bool: + if self.time_expiry is None: + return False + return _as_utc(now) >= _as_utc(self.time_expiry) + + class LocalUser(BaseModel): model_config = ConfigDict(populate_by_name=True) @@ -73,6 +116,17 @@ class LocalUser(BaseModel): pass_hash: str = "" email: str | None = None assume_anonymous: bool = False + keys: list[HttpKey] = Field(default_factory=list) + + @field_validator("keys") + @classmethod + def _unique_key_ids(cls, keys: list[HttpKey]) -> list[HttpKey]: + seen: set[str] = set() + for key in keys: + if key.id in seen: + raise ValueError(f"duplicate http key id: {key.id}") + seen.add(key.id) + return keys @field_validator("email", mode="before") @classmethod @@ -168,10 +222,11 @@ def _migrate_plaintext_passwords(path: Path, config: AuthConfig) -> AuthConfig: updated_users.append(entry) continue updated_users.append( - LocalUser( - user=entry.user, - pass_hash=hash_password(entry.password, config_salt=config.salt), - email=entry.email, + entry.model_copy( + update={ + "password": None, + "pass_hash": hash_password(entry.password, config_salt=config.salt), + } ) ) changed = True diff --git a/pygitweb/src/pygitweb/auth_test.py b/pygitweb/src/pygitweb/auth_test.py index 1b996c3..7fc0e76 100644 --- a/pygitweb/src/pygitweb/auth_test.py +++ b/pygitweb/src/pygitweb/auth_test.py @@ -3,6 +3,7 @@ import json import os import sys from collections.abc import Generator +from datetime import UTC, datetime from pathlib import Path from unittest.mock import AsyncMock, patch from urllib.parse import parse_qs, urlparse @@ -12,9 +13,10 @@ import pytest from fastapi import FastAPI, HTTPException, Request from fastapi.testclient import TestClient from httpx import ASGITransport +from pydantic import ValidationError from pygitweb.auth import ACCESS_TOKEN_COOKIE_NAME -from pygitweb.auth_config import AuthConfig, LocalUser, init_auth_config, write_auth_config +from pygitweb.auth_config import AuthConfig, HttpKey, LocalUser, init_auth_config, write_auth_config from pygitweb.auth_oauth import ( OAUTH_STATE_COOKIE_NAME, _exchange_code, @@ -293,6 +295,138 @@ def test_auth_config_migrates_plaintext_pass_to_pass_hash(tmp_path: Path) -> Non assert "pass_hash" in user_entry +def test_auth_config_keeps_hashed_http_key(tmp_path: Path) -> None: + cfg_path = tmp_path / "auth.json" + key_hash = hash_password("plain-key-value") + cfg_path.write_text( + json.dumps({ + "auth_mode": "local", + "salt": "0123456789abcdef", + "local_users": [ + { + "user": "admin", + "pass": "secret1234", + "email": "admin@example.com", + "assume_anonymous": True, + "keys": [ + { + "id": "laptop", + "secret": key_hash, + "readonly": True, + "time_issued": "2026-10-08T12:00:00Z", + "time_expiry": "2027-10-08T12:00:00Z", + } + ], + } + ], + }), + encoding="utf-8", + ) + if sys.platform != "win32": + os.chmod(cfg_path, 0o600) + s = Settings(AUTH=True, AUTH_CONFIG=str(cfg_path)) + cfg = init_auth_config(s) + user = cfg.local_users[0] + assert user.password is None + assert user.assume_anonymous is True + assert user.email == "admin@example.com" + assert verify_password("secret1234", user.pass_hash, config_salt=cfg.salt) + assert len(user.keys) == 1 + key = user.keys[0] + assert key.id == "laptop" + assert key.readonly is True + assert key.time_issued == datetime(2026, 10, 8, 12, 0, tzinfo=UTC) + assert key.time_expiry == datetime(2027, 10, 8, 12, 0, tzinfo=UTC) + assert key.secret == key_hash + assert verify_password("plain-key-value", key.secret) + on_disk = json.loads(cfg_path.read_text(encoding="utf-8")) + stored = on_disk["local_users"][0] + assert "pass" not in stored + assert stored["keys"][0]["secret"] == key_hash + assert stored["keys"][0]["readonly"] is True + assert stored["keys"][0]["time_expiry"] == "2027-10-08T12:00:00Z" + + +def test_auth_config_rejects_plaintext_http_key(tmp_path: Path) -> None: + cfg_path = tmp_path / "auth.json" + original = json.dumps({ + "auth_mode": "local", + "local_users": [ + { + "user": "admin", + "pass": "secret1234", + "keys": [ + { + "id": "laptop", + "secret": "plain-key-value", + "time_issued": "2026-10-08T12:00:00Z", + } + ], + } + ], + }) + cfg_path.write_text(original, encoding="utf-8") + if sys.platform != "win32": + os.chmod(cfg_path, 0o600) + s = Settings(AUTH=True, AUTH_CONFIG=str(cfg_path)) + with pytest.raises(ValidationError, match="scrypt password hash"): + init_auth_config(s) + assert cfg_path.read_text(encoding="utf-8") == original + + +def test_http_key_rejects_duplicate_ids() -> None: + issued = datetime(2026, 10, 8, tzinfo=UTC) + with pytest.raises(ValidationError, match="duplicate http key id"): + LocalUser( + user="admin", + password="secret", + keys=[ + HttpKey(id="laptop", secret=hash_password("one"), readonly=False, time_issued=issued), + HttpKey(id="laptop", secret=hash_password("two"), readonly=True, time_issued=issued), + ], + ) + + +def test_http_key_strips_id_and_secret_and_rejects_blanks() -> None: + issued = datetime(2026, 10, 8, tzinfo=UTC) + key_hash = hash_password("secret") + key = HttpKey(id=" laptop ", secret=f" {key_hash} ", time_issued=issued) + assert key.id == "laptop" + assert key.secret == key_hash + assert key.time_expiry is None + with pytest.raises(ValidationError, match="non-empty"): + HttpKey(id=" ", secret=key_hash, time_issued=issued) + with pytest.raises(ValidationError, match="non-empty"): + HttpKey(id="laptop", secret=" ", time_issued=issued) + with pytest.raises(ValidationError, match="scrypt password hash"): + HttpKey(id="laptop", secret="plaintext", time_issued=issued) + with pytest.raises(ValidationError): + HttpKey.model_validate({"id": 1, "secret": key_hash, "time_issued": issued.isoformat()}) + + +def test_http_key_rejects_expiry_before_issue() -> None: + issued = datetime(2026, 10, 8, tzinfo=UTC) + key_hash = hash_password("secret") + with pytest.raises(ValidationError, match="time_expiry is before time_issued"): + HttpKey( + id="laptop", + secret=key_hash, + time_issued=issued, + time_expiry=datetime(2026, 10, 7, tzinfo=UTC), + ) + naive_expiry = datetime(2026, 10, 7, 0, 0) + with pytest.raises(ValidationError, match="time_expiry is before time_issued"): + HttpKey(id="laptop", secret=key_hash, time_issued=issued, time_expiry=naive_expiry) + active = HttpKey( + id="laptop", + secret=key_hash, + time_issued=issued, + time_expiry=datetime(2026, 10, 9), + ) + assert active.is_expired(datetime(2026, 10, 8, 12, tzinfo=UTC)) is False + assert active.is_expired(datetime(2026, 10, 9)) is True + + @pytest.mark.skipif(sys.platform == "win32", reason="Unix file modes") def test_auth_config_rejects_insecure_permissions(tmp_path: Path) -> None: from pygitweb.auth_config import AuthConfigPermissionError, init_auth_config, write_auth_config diff --git a/pygitweb/src/pygitweb/config.py b/pygitweb/src/pygitweb/config.py index b5fb416..0636f29 100644 --- a/pygitweb/src/pygitweb/config.py +++ b/pygitweb/src/pygitweb/config.py @@ -122,6 +122,12 @@ class Settings(BaseSettings): AUTH_CONFIG: str = "" SETTINGS_CONFIG: str = "" ENABLED_PLUGINS: list[str] = Field(default_factory=list) + API_KEY_MAX_COUNT: int = Field(default=2, ge=0) + API_KEY_EXPIRY_DAYS: int = Field( + default=0, + ge=0, + description="Lifetime of a newly issued API key, in days. 0 means the key does not expire.", + ) @field_validator("AUTH", mode="before") @classmethod diff --git a/pygitweb/src/pygitweb/password_hash.py b/pygitweb/src/pygitweb/password_hash.py index f4949b7..ce24fe9 100644 --- a/pygitweb/src/pygitweb/password_hash.py +++ b/pygitweb/src/pygitweb/password_hash.py @@ -43,6 +43,21 @@ def hash_password(password: str, *, config_salt: str = "") -> str: return f"{_ALGORITHM}${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}${salt_b64}${key_b64}" +def is_password_hash(value: str) -> bool: + parts = value.split("$") + if len(parts) != 6 or parts[0] != _ALGORITHM: + return False + try: + int(parts[1]) + int(parts[2]) + int(parts[3]) + base64.b64decode(parts[4], validate=True) + base64.b64decode(parts[5], validate=True) + except (ValueError, TypeError): + return False + return True + + def verify_password(password: str, pass_hash: str, *, config_salt: str = "") -> bool: if not pass_hash: return False diff --git a/pygitweb/src/pygitweb/password_hash_test.py b/pygitweb/src/pygitweb/password_hash_test.py index 327b692..6c6bc51 100644 --- a/pygitweb/src/pygitweb/password_hash_test.py +++ b/pygitweb/src/pygitweb/password_hash_test.py @@ -1,4 +1,4 @@ -from pygitweb.password_hash import hash_password, verify_password +from pygitweb.password_hash import hash_password, is_password_hash, verify_password def test_hash_and_verify_roundtrip() -> None: @@ -13,6 +13,13 @@ def test_verify_legacy_hash_without_config_salt() -> None: assert verify_password("legacy", stored, config_salt="0123456789abcdef") +def test_is_password_hash_rejects_malformed_values() -> None: + assert is_password_hash(hash_password("secret")) + assert not is_password_hash("plaintext") + assert not is_password_hash("scrypt$nope$8$1$aaaa$bbbb") + assert not is_password_hash("scrypt$131072$8$1$!!!$!!!") + + def test_verify_rejects_weak_parameters() -> None: stored = hash_password("x") parts = stored.split("$") diff --git a/pygitweb/src/pygitweb/settings.py b/pygitweb/src/pygitweb/settings.py index 84046f6..a8de0b9 100644 --- a/pygitweb/src/pygitweb/settings.py +++ b/pygitweb/src/pygitweb/settings.py @@ -52,6 +52,12 @@ PYGITWEB_SETTINGS: list[tuple[str, type, str]] = [ ("STRICT_EXPORT", bool, "When true, require export_ok (or path match) to list projects."), ("GIT", str, "Path to the git executable (e.g. for maintenance)."), ("MAXLOAD", float, "Max load average; 503 when exceeded. Empty to disable."), + ("API_KEY_MAX_COUNT", int, "Maximum Smart HTTP API keys each local user may hold."), + ( + "API_KEY_EXPIRY_DAYS", + int, + "Lifetime of a newly issued API key, in days. 0 means the key does not expire.", + ), ] @@ -327,6 +333,10 @@ async def settings_pygitweb_submit(request: Request): elif expected_type is float: with suppress(ValueError): setattr(settings, name, float(raw) if raw else None) + elif expected_type is int: + if raw: + with suppress(ValueError): + setattr(settings, name, int(raw)) elif raw: setattr(settings, name, raw) return RedirectResponse(url="/settings/pygitweb", status_code=303) diff --git a/pygitweb/src/pygitweb/smart_http.py b/pygitweb/src/pygitweb/smart_http.py index f54787a..0daa29f 100644 --- a/pygitweb/src/pygitweb/smart_http.py +++ b/pygitweb/src/pygitweb/smart_http.py @@ -5,6 +5,9 @@ from __future__ import annotations import gzip import os import subprocess +import sys +import threading +from datetime import UTC, datetime from pathlib import Path from typing import Annotated, Literal @@ -12,7 +15,16 @@ import pygit2 from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status from fastapi.responses import Response -from pygitweb.auth import access_token_from_request, ensure_read_project, ensure_write_project +from pygitweb.auth import ( + ACCESS_TOKEN_COOKIE_NAME, + SmartHttpCredentials, + can_read_project, + load_smart_http_credentials, + principal_can_read_project, + principal_can_write_project, + raise_for_smart_http_credentials, + unauthorized_smart_http, +) from pygitweb.config import settings from pygitweb.dependencies import project_visible_in_list, require_valid_project from pygitweb.validation import is_valid_project @@ -21,6 +33,9 @@ SmartHttpService = Literal["git-upload-pack", "git-receive-pack"] http_router = APIRouter(prefix="/http", tags=["smart_http"]) +_SMART_HTTP_LOG_LOCK = threading.Lock() +_SMART_HTTP_LOG_MODE = 0o600 + _SERVICE_CONTENT_TYPES: dict[SmartHttpService, tuple[str, str]] = { "git-upload-pack": ( "application/x-git-upload-pack-advertisement", @@ -76,27 +91,69 @@ def require_http_project(project: str) -> str: return require_valid_project(project) -def require_readable_http_project( - project: str, - token: Annotated[str | None, Depends(access_token_from_request)], -) -> str: - validated = require_http_project(project) - ensure_read_project(validated, token) - return validated +ValidatedHttpProject = Annotated[str, Depends(require_http_project)] -def require_writable_http_project( - project: str, - token: Annotated[str | None, Depends(access_token_from_request)], -) -> str: - validated = require_http_project(project) - ensure_write_project(validated, token) - return validated +def smart_http_log_path() -> Path: + return Path.home() / ".pygitweb" / "smart_http.log" -ValidatedHttpProject = Annotated[str, Depends(require_http_project)] -ValidatedReadableHttpProject = Annotated[str, Depends(require_readable_http_project)] -ValidatedWritableHttpProject = Annotated[str, Depends(require_writable_http_project)] +def _request_ip(request: Request) -> str: + client = request.client + if client is None or not client.host: + return "-" + return client.host.replace("\r", "").replace("\n", "") + + +def _log_token(value: str) -> str: + return value.replace("\r", "").replace("\n", "") or "-" + + +def append_smart_http_log(ip: str, actor: str, *, project: str, operation: str) -> None: + stamp = datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") + safe_actor = actor.replace("\r", "").replace("\n", "") + line = f"{stamp} {_log_token(ip)} {_log_token(project)} {_log_token(operation)}" + if safe_actor: + line = f"{line} {safe_actor}" + path = smart_http_log_path() + path.parent.mkdir(parents=True, exist_ok=True) + with _SMART_HTTP_LOG_LOCK: + created = not path.exists() + with path.open("a", encoding="utf-8", newline="\n") as handle: + handle.write(f"{line}\n") + if created and sys.platform != "win32": + os.chmod(path, _SMART_HTTP_LOG_MODE) + + +def _begin_smart_http(request: Request, project: str, *, write: bool, operation: str) -> None: + creds = load_smart_http_credentials( + request.headers.get("authorization"), + request.cookies.get(ACCESS_TOKEN_COOKIE_NAME), + project=project, + write=write, + ) + append_smart_http_log(_request_ip(request), creds.actor, project=project, operation=operation) + raise_for_smart_http_credentials(creds) + _authorize_smart_http(project, creds, write=write) + + +def _authorize_smart_http(project: str, creds: SmartHttpCredentials, *, write: bool) -> None: + if not creds.required: + return + principal = creds.principal + if principal is None: + if not write and can_read_project(project, None): + return + raise unauthorized_smart_http() + if write and creds.readonly: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Read-only key") + if write: + if principal_can_write_project(project, principal): + return + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions") + if principal_can_read_project(project, principal): + return + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such project") def _repo_cwd(project: str) -> Path: @@ -164,12 +221,14 @@ def _smart_http_cache_headers() -> dict[str, str]: def smart_http_info_refs( project: ValidatedHttpProject, service: Annotated[SmartHttpService, Query(description="Git Smart HTTP service name")], - token: Annotated[str | None, Depends(access_token_from_request)], + request: Request, ) -> Response: - if service == "git-receive-pack": - ensure_write_project(project, token) - else: - ensure_read_project(project, token) + _begin_smart_http( + request, + project, + write=service == "git-receive-pack", + operation=f"info/refs?service={service}", + ) body = _service_advertisement(service, project) advertise_type, _ = _SERVICE_CONTENT_TYPES[service] return Response( @@ -181,11 +240,12 @@ def smart_http_info_refs( @http_router.post("/{project:path}/git-upload-pack") async def smart_http_upload_pack( - project: ValidatedReadableHttpProject, + project: ValidatedHttpProject, request: Request, content_encoding: Annotated[str | None, Header(alias="Content-Encoding")] = None, git_protocol: Annotated[str | None, Header(alias="Git-Protocol")] = None, ) -> Response: + _begin_smart_http(request, project, write=False, operation="git-upload-pack") raw_body = await request.body() body = _decode_request_body(raw_body, content_encoding) result = _run_git_service( @@ -201,11 +261,12 @@ async def smart_http_upload_pack( @http_router.post("/{project:path}/git-receive-pack") async def smart_http_receive_pack( - project: ValidatedWritableHttpProject, + project: ValidatedHttpProject, request: Request, content_encoding: Annotated[str | None, Header(alias="Content-Encoding")] = None, git_protocol: Annotated[str | None, Header(alias="Git-Protocol")] = None, ) -> Response: + _begin_smart_http(request, project, write=True, operation="git-receive-pack") raw_body = await request.body() body = _decode_request_body(raw_body, content_encoding) result = _run_git_service( diff --git a/pygitweb/src/pygitweb/smart_http_test.py b/pygitweb/src/pygitweb/smart_http_test.py index 2ffe399..32649fb 100644 --- a/pygitweb/src/pygitweb/smart_http_test.py +++ b/pygitweb/src/pygitweb/smart_http_test.py @@ -1,11 +1,18 @@ from __future__ import annotations +import base64 import gzip +import os +import re import socket +import stat import subprocess +import sys import threading import time from collections.abc import Generator +from contextlib import contextmanager +from datetime import UTC, datetime, timedelta from pathlib import Path from unittest.mock import patch @@ -14,12 +21,28 @@ import pytest import uvicorn from fastapi.testclient import TestClient -from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.auth import ACCESS_TOKEN_COOKIE_NAME +from pygitweb.auth_config import AuthConfig, HttpKey, LocalUser from pygitweb.config import settings from pygitweb.conftest import set_client_session from pygitweb.main import app +from pygitweb.password_hash import hash_password from pygitweb.permissions import PermissionsMap -from pygitweb.smart_http import _decode_request_body, _pkt_flush, _pkt_line, _service_advertisement +from pygitweb.sessions import create_session +from pygitweb.smart_http import ( + _decode_request_body, + _pkt_flush, + _pkt_line, + _service_advertisement, + append_smart_http_log, + smart_http_log_path, +) + +_KEY_ISSUED = datetime(2026, 10, 8, 12, 0, tzinfo=UTC) +_WRITE_KEY = "read-write-key" +_READ_KEY = "read-only-key" +_WRITE_KEY_HASH = hash_password(_WRITE_KEY) +_READ_KEY_HASH = hash_password(_READ_KEY) _LOCAL_ADMIN = AuthConfig( auth_mode="local", @@ -28,6 +51,47 @@ _LOCAL_ADMIN = AuthConfig( ) +def _user_with_keys(*, secret: str | None = None, salt: str = "") -> AuthConfig: + if salt: + write_secret = secret if secret is not None else hash_password(_WRITE_KEY, config_salt=salt) + read_secret = hash_password(_READ_KEY, config_salt=salt) + else: + write_secret = secret if secret is not None else _WRITE_KEY_HASH + read_secret = _READ_KEY_HASH + return AuthConfig( + auth_mode="local", + salt=salt, + local_users=[ + LocalUser( + user="alice", + password="secret", + email="alice@example.com", + keys=[ + HttpKey(id="laptop", secret=write_secret, readonly=False, time_issued=_KEY_ISSUED), + HttpKey( + id="ci", + secret=read_secret, + readonly=True, + time_issued=_KEY_ISSUED, + ), + ], + ) + ], + oauth_permissions=PermissionsMap.default_access(), + ) + + +@contextmanager +def _auth_enabled(cfg: AuthConfig) -> Generator[None, None, None]: + with ( + patch.object(settings, "AUTH", True), + patch("pygitweb.auth_config.auth_config", cfg), + patch("pygitweb.auth.auth_config", cfg), + patch("pygitweb.main.auth_config", cfg), + ): + yield + + @pytest.fixture def local_auth_config() -> Generator[AuthConfig, None, None]: with ( @@ -38,17 +102,46 @@ def local_auth_config() -> Generator[AuthConfig, None, None]: yield _LOCAL_ADMIN -def _run_git(repo_dir: Path, *args: str) -> str: +def _run_git(repo_dir: Path, *args: str, git_config: tuple[str, ...] = ()) -> str: completed = subprocess.run( - [settings.GIT, *args], + [settings.GIT, *git_config, *args], cwd=repo_dir, capture_output=True, text=True, check=True, + env={**os.environ, "GIT_TERMINAL_PROMPT": "0"}, ) return completed.stdout.strip() +@contextmanager +def _http_server() -> Generator[int, None, None]: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error", ws="none") + server = uvicorn.Server(config) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + deadline = time.time() + 5.0 + while not server.started and time.time() < deadline: + time.sleep(0.01) + if not server.started: + server.should_exit = True + thread.join(timeout=5.0) + raise RuntimeError("smart HTTP server did not start") + try: + yield port + finally: + server.should_exit = True + thread.join(timeout=5.0) + + +def _key_git_config() -> tuple[str, ...]: + helper = f"!f() {{ echo username=alice; echo password={_WRITE_KEY}; }}; f" + return ("-c", "credential.helper=", "-c", f"credential.helper={helper}") + + def _seed_bare_repo(root: Path, bare_dir: Path, message: str = "initial") -> None: work_dir = root / "seed-work" work_dir.mkdir() @@ -66,6 +159,7 @@ def _seed_bare_repo(root: Path, bare_dir: Path, message: str = "initial") -> Non def http_repo_env(tmp_path_factory: pytest.TempPathFactory) -> Generator[dict[str, str], None, None]: root = tmp_path_factory.mktemp("smart-http") repo_dir = root / "demo" + log_path = root / "smart_http.log" pygit2.init_repository(str(repo_dir), bare=True) _seed_bare_repo(root, repo_dir) with ( @@ -76,8 +170,14 @@ def http_repo_env(tmp_path_factory: pytest.TempPathFactory) -> Generator[dict[st patch.object(settings, "EXPORT_OK", ""), patch.object(settings, "LIST_ALL", True), patch.object(settings, "AUTH", False), + patch("pygitweb.smart_http.smart_http_log_path", return_value=log_path), ): - yield {"project": "demo", "repo_dir": str(repo_dir), "root": str(root)} + yield { + "project": "demo", + "repo_dir": str(repo_dir), + "root": str(root), + "log": str(log_path), + } class TestSmartHttpHelpers: @@ -192,29 +292,127 @@ class TestSmartHttpRoutes: assert resp.status_code == 200 assert resp.headers["content-type"].startswith("application/x-git-receive-pack-advertisement") + def test_fetch_requires_auth_when_enabled( + self, + http_repo_env: dict[str, str], + ) -> None: + client = TestClient(app) + with _auth_enabled(_user_with_keys()): + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + ) + assert resp.status_code == 401 + assert "Basic" in resp.headers["www-authenticate"] + + def test_http_key_allows_fetch_and_push( + self, + http_repo_env: dict[str, str], + ) -> None: + client = TestClient(app) + with _auth_enabled(_user_with_keys()): + fetch = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _WRITE_KEY), + ) + push = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-receive-pack"}, + auth=("Alice@Example.com", _WRITE_KEY), + ) + assert fetch.status_code == 200 + assert push.status_code == 200 + + def test_readonly_key_can_fetch_but_not_push( + self, + http_repo_env: dict[str, str], + ) -> None: + client = TestClient(app) + with _auth_enabled(_user_with_keys()): + fetch = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice@example.com", _READ_KEY), + ) + push = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-receive-pack"}, + auth=("alice", _READ_KEY), + ) + post_push = client.post( + f"/http/{http_repo_env['project']}/git-receive-pack", + content=b"0000", + auth=("alice", _READ_KEY), + ) + assert fetch.status_code == 200 + assert push.status_code == 403 + assert post_push.status_code == 403 + + def test_login_password_is_not_an_http_key( + self, + http_repo_env: dict[str, str], + ) -> None: + client = TestClient(app) + with _auth_enabled(_user_with_keys()): + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", "secret"), + ) + web = client.get("/users/me", auth=("alice", _WRITE_KEY)) + assert resp.status_code == 401 + assert web.status_code == 401 + + def test_hashed_http_key_allows_fetch( + self, + http_repo_env: dict[str, str], + ) -> None: + salt = "0123456789abcdef" + cfg = _user_with_keys(secret=hash_password(_WRITE_KEY, config_salt=salt), salt=salt) + client = TestClient(app) + with _auth_enabled(cfg): + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _WRITE_KEY), + ) + wrong = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", "not-the-key"), + ) + assert resp.status_code == 200 + assert wrong.status_code == 401 + + def test_http_key_without_project_permission( + self, + http_repo_env: dict[str, str], + ) -> None: + cfg = _user_with_keys() + cfg = cfg.model_copy(update={"oauth_permissions": PermissionsMap.empty()}) + client = TestClient(app) + with ( + _auth_enabled(cfg), + patch("pygitweb.auth.git_get_project_owner", return_value=None), + ): + fetch = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _WRITE_KEY), + ) + push = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-receive-pack"}, + auth=("alice", _WRITE_KEY), + ) + assert fetch.status_code == 404 + assert push.status_code == 403 + def test_clone_and_push_over_http(self, http_repo_env: dict[str, str]) -> None: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.bind(("127.0.0.1", 0)) - port = sock.getsockname()[1] - - config = uvicorn.Config( - app, - host="127.0.0.1", - port=port, - log_level="error", - ws="none", - ) - server = uvicorn.Server(config) - thread = threading.Thread(target=server.run, daemon=True) - thread.start() - deadline = time.time() + 5.0 - while not server.started and time.time() < deadline: - time.sleep(0.01) - assert server.started - - clone_dir = Path(http_repo_env["root"]) / "clone" - remote_url = f"http://127.0.0.1:{port}/http/{http_repo_env['project']}.git" - try: + with _http_server() as port: + clone_dir = Path(http_repo_env["root"]) / "clone" + remote_url = f"http://127.0.0.1:{port}/http/{http_repo_env['project']}.git" _run_git(Path(http_repo_env["root"]), "clone", remote_url, str(clone_dir)) _run_git(clone_dir, "config", "user.name", "tester") _run_git(clone_dir, "config", "user.email", "tester@example.com") @@ -227,6 +425,457 @@ class TestSmartHttpRoutes: assert commit.message.strip() == "add new" tree = commit.tree assert tree["new.txt"].data.decode("utf-8") == "pushed\n" - finally: - server.should_exit = True - thread.join(timeout=5.0) + + def test_expired_http_key_is_rejected_and_unexpired_key_still_works( + self, + http_repo_env: dict[str, str], + ) -> None: + now = datetime.now(UTC) + cfg = _user_with_keys() + user = cfg.local_users[0] + cfg = cfg.model_copy( + update={ + "local_users": [ + user.model_copy( + update={ + "keys": [ + HttpKey( + id="old", + secret=_WRITE_KEY_HASH, + readonly=False, + time_issued=_KEY_ISSUED, + time_expiry=now - timedelta(seconds=1), + ), + HttpKey( + id="ci", + secret=_READ_KEY_HASH, + readonly=True, + time_issued=_KEY_ISSUED, + time_expiry=now + timedelta(days=1), + ), + ] + } + ) + ] + } + ) + client = TestClient(app) + with _auth_enabled(cfg): + expired = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _WRITE_KEY), + ) + current = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _READ_KEY), + ) + assert expired.status_code == 401 + assert current.status_code == 200 + + def test_http_key_rejects_unknown_user_and_user_without_email( + self, + http_repo_env: dict[str, str], + ) -> None: + cfg = AuthConfig( + auth_mode="local", + local_users=[ + LocalUser( + user="alice", + password="secret", + email="alice@example.com", + keys=[HttpKey(id="laptop", secret=_WRITE_KEY_HASH, time_issued=_KEY_ISSUED)], + ), + LocalUser( + user="bob", + password="secret", + keys=[HttpKey(id="bob-key", secret=hash_password("bob-secret"), time_issued=_KEY_ISSUED)], + ), + ], + oauth_permissions=PermissionsMap.default_access(), + ) + client = TestClient(app) + with _auth_enabled(cfg): + wrong_user = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("carol", _WRITE_KEY), + ) + blank_user = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + headers={"Authorization": f"Basic {base64.b64encode(b' :' + _WRITE_KEY.encode()).decode()}"}, + ) + assert wrong_user.status_code == 401 + assert blank_user.status_code == 401 + + def test_malformed_and_non_basic_authorization_rejected( + self, + http_repo_env: dict[str, str], + ) -> None: + client = TestClient(app) + url = f"/http/{http_repo_env['project']}/info/refs" + params = {"service": "git-upload-pack"} + with _auth_enabled(_user_with_keys()): + not_basic = client.get(url, params=params, headers={"Authorization": "Digest abc"}) + empty_param = client.get(url, params=params, headers={"Authorization": "Basic"}) + bad_b64 = client.get(url, params=params, headers={"Authorization": "Basic @@@@"}) + no_colon = client.get( + url, + params=params, + headers={"Authorization": f"Basic {base64.b64encode(b'alice').decode()}"}, + ) + bad_bearer = client.get(url, params=params, headers={"Authorization": "Bearer not-a-session"}) + client.cookies.set(ACCESS_TOKEN_COOKIE_NAME, "not-a-session") + bad_cookie = client.get(url, params=params) + client.cookies.clear() + bad_utf8 = client.get( + url, + params=params, + headers={"Authorization": f"Basic {base64.b64encode(b'\xff\xfe:secret').decode()}"}, + ) + empty_secret = client.get( + url, + params=params, + headers={"Authorization": f"Basic {base64.b64encode(b'alice:').decode()}"}, + ) + padded_secret = client.get( + url, + params=params, + auth=(" alice ", f" {_WRITE_KEY} "), + ) + assert not_basic.status_code == 401 + assert empty_param.status_code == 401 + assert bad_b64.status_code == 401 + assert no_colon.status_code == 401 + assert bad_utf8.status_code == 401 + assert empty_secret.status_code == 401 + assert padded_secret.status_code == 200 + assert bad_bearer.status_code == 401 + assert bad_cookie.status_code == 401 + + def test_bearer_session_allows_fetch(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + with _auth_enabled(_user_with_keys()): + token = create_session("alice", email="alice@example.com", auth_method="local") + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + headers={"Authorization": f"Bearer {token}"}, + ) + assert resp.status_code == 200 + + def test_keys_only_user_can_fetch_when_local_password_absent( + self, + http_repo_env: dict[str, str], + ) -> None: + cfg = AuthConfig( + auth_mode="local", + local_users=[ + LocalUser( + user="alice", + email="alice@example.com", + keys=[HttpKey(id="laptop", secret=_WRITE_KEY_HASH, time_issued=_KEY_ISSUED)], + ) + ], + oauth_permissions=PermissionsMap.default_access(), + ) + client = TestClient(app) + with _auth_enabled(cfg): + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _WRITE_KEY), + ) + assert resp.status_code == 200 + + def test_unconfigured_auth_without_keys_is_unavailable( + self, + http_repo_env: dict[str, str], + ) -> None: + cfg = AuthConfig(auth_mode="local", local_users=[], oauth_permissions=PermissionsMap.empty()) + client = TestClient(app) + with _auth_enabled(cfg): + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + ) + assert resp.status_code == 503 + + def test_anonymous_reader_can_fetch_but_not_push( + self, + http_repo_env: dict[str, str], + ) -> None: + cfg = _user_with_keys() + user = cfg.local_users[0].model_copy(update={"assume_anonymous": True}) + cfg = cfg.model_copy(update={"local_users": [user]}) + client = TestClient(app) + with _auth_enabled(cfg): + fetch = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + ) + push = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-receive-pack"}, + ) + assert fetch.status_code == 200 + assert push.status_code == 401 + + def test_clone_and_push_with_http_key(self, http_repo_env: dict[str, str]) -> None: + with _auth_enabled(_user_with_keys()), _http_server() as port: + clone_dir = Path(http_repo_env["root"]) / "keyed-clone" + remote_url = f"http://127.0.0.1:{port}/http/{http_repo_env['project']}.git" + git_config = _key_git_config() + _run_git(Path(http_repo_env["root"]), "clone", remote_url, str(clone_dir), git_config=git_config) + _run_git(clone_dir, "config", "user.name", "tester") + _run_git(clone_dir, "config", "user.email", "tester@example.com") + (clone_dir / "from-key.txt").write_text("keyed\n", encoding="utf-8") + _run_git(clone_dir, "add", "from-key.txt") + _run_git(clone_dir, "commit", "-m", "add keyed") + _run_git(clone_dir, "push", "origin", "HEAD", git_config=git_config) + repo = pygit2.Repository(http_repo_env["repo_dir"]) + commit = repo.head.peel() + assert commit.message.strip() == "add keyed" + assert commit.tree["from-key.txt"].data.decode("utf-8") == "keyed\n" + + +_LOG_LINE = re.compile(r"^(?P\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z) (?P.+)$") + + +def _log_lines(http_repo_env: dict[str, str]) -> list[str]: + path = Path(http_repo_env["log"]) + if not path.is_file(): + return [] + return path.read_text(encoding="utf-8").splitlines() + + +def _assert_log_line(line: str, actor: str | None, *, project: str, operation: str) -> None: + match = _LOG_LINE.fullmatch(line) + assert match is not None, line + identity = "" if actor is None else f" {actor}" + assert match.group("rest") == f"testclient {project} {operation}{identity}" + + +class TestSmartHttpLogFile: + def test_default_path_is_under_pygitweb_home(self, tmp_path: Path) -> None: + with patch("pygitweb.smart_http.Path.home", return_value=tmp_path): + assert smart_http_log_path() == tmp_path / ".pygitweb" / "smart_http.log" + + def test_append_keeps_earlier_lines_and_restricts_mode(self, tmp_path: Path) -> None: + path = tmp_path / ".pygitweb" / "smart_http.log" + with patch("pygitweb.smart_http.smart_http_log_path", return_value=path): + append_smart_http_log("203.0.113.8", "alice@example.com abcd", project="demo", operation="git-upload-pack") + append_smart_http_log("203.0.113.9", "INVALID_KEY", project="demo", operation="git-receive-pack") + lines = path.read_text(encoding="utf-8").splitlines() + assert lines[0].endswith(" 203.0.113.8 demo git-upload-pack alice@example.com abcd") + assert lines[1].endswith(" 203.0.113.9 demo git-receive-pack INVALID_KEY") + assert path.read_bytes().endswith(b"INVALID_KEY\n") + if sys.platform != "win32": + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +class TestSmartHttpAccessLog: + def test_each_route_logs_owner_email_and_key_tail(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + project = http_repo_env["project"] + with _auth_enabled(_user_with_keys()): + before = len(_log_lines(http_repo_env)) + refs = client.get( + f"/http/{project}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _WRITE_KEY), + ) + upload = client.post( + f"/http/{project}/git-upload-pack", + content=b"0000", + auth=(" alice ", f" {_WRITE_KEY} "), + ) + push_refs = client.get( + f"/http/{project}/info/refs", + params={"service": "git-receive-pack"}, + auth=("Alice@Example.com", _WRITE_KEY), + ) + receive = client.post( + f"/http/{project}/git-receive-pack", + content=b"0000", + auth=("alice", _WRITE_KEY), + ) + assert refs.status_code == 200 + assert push_refs.status_code == 200 + assert upload.status_code != 401 + assert receive.status_code != 401 + lines = _log_lines(http_repo_env)[before:] + assert len(lines) == 4 + operations = ( + "info/refs?service=git-upload-pack", + "git-upload-pack", + "info/refs?service=git-receive-pack", + "git-receive-pack", + ) + for line, operation in zip(lines, operations, strict=True): + _assert_log_line(line, "alice@example.com -key", project=project, operation=operation) + assert _WRITE_KEY not in line + + def test_broken_key_logs_invalid_key_and_expired_key_marks_tail(self, http_repo_env: dict[str, str]) -> None: + now = datetime.now(UTC) + cfg = _user_with_keys() + user = cfg.local_users[0] + cfg = cfg.model_copy( + update={ + "local_users": [ + user.model_copy( + update={ + "keys": [ + HttpKey( + id="old", + secret=_WRITE_KEY_HASH, + readonly=False, + time_issued=_KEY_ISSUED, + time_expiry=now - timedelta(seconds=1), + ), + HttpKey( + id="ci", + secret=_READ_KEY_HASH, + readonly=True, + time_issued=_KEY_ISSUED, + time_expiry=now + timedelta(days=1), + ), + ] + } + ) + ] + } + ) + client = TestClient(app) + project = http_repo_env["project"] + url = f"/http/{project}/info/refs" + params = {"service": "git-upload-pack"} + with _auth_enabled(cfg): + before = len(_log_lines(http_repo_env)) + expired = client.get(url, params=params, auth=("alice", _WRITE_KEY)) + broken = client.get(url, params=params, auth=("alice", "not-the-key")) + malformed = client.get(url, params=params, headers={"Authorization": "Basic @@@@"}) + current = client.get(url, params=params, auth=("alice", _READ_KEY)) + assert expired.status_code == 401 + assert broken.status_code == 401 + assert malformed.status_code == 401 + assert current.status_code == 200 + lines = _log_lines(http_repo_env)[before:] + assert len(lines) == 4 + operation = "info/refs?service=git-upload-pack" + _assert_log_line(lines[0], "alice@example.com -key (EXPIRED)", project=project, operation=operation) + _assert_log_line(lines[1], "INVALID_KEY", project=project, operation=operation) + _assert_log_line(lines[2], "INVALID_KEY", project=project, operation=operation) + _assert_log_line(lines[3], "alice@example.com -key", project=project, operation=operation) + logged = "\n".join(lines) + assert _WRITE_KEY not in logged + assert _READ_KEY not in logged + + def test_public_fetch_logs_username_instead_of_credentials(self, http_repo_env: dict[str, str]) -> None: + cfg = _user_with_keys() + user = cfg.local_users[0].model_copy(update={"assume_anonymous": True}) + cfg = cfg.model_copy(update={"local_users": [user]}) + client = TestClient(app) + project = http_repo_env["project"] + with _auth_enabled(cfg): + before = len(_log_lines(http_repo_env)) + fetch = client.get( + f"/http/{project}/info/refs", + params={"service": "git-upload-pack"}, + ) + upload = client.post(f"/http/{project}/git-upload-pack", content=b"0000") + push = client.get( + f"/http/{project}/info/refs", + params={"service": "git-receive-pack"}, + ) + keyed = client.get( + f"/http/{project}/info/refs", + params={"service": "git-upload-pack"}, + auth=("alice", _WRITE_KEY), + ) + assert fetch.status_code == 200 + assert upload.status_code != 401 + assert push.status_code == 401 + assert keyed.status_code == 200 + lines = _log_lines(http_repo_env)[before:] + assert len(lines) == 4 + _assert_log_line(lines[0], "alice", project=project, operation="info/refs?service=git-upload-pack") + _assert_log_line(lines[1], "alice", project=project, operation="git-upload-pack") + _assert_log_line(lines[2], None, project=project, operation="info/refs?service=git-receive-pack") + _assert_log_line( + lines[3], "alice@example.com -key", project=project, operation="info/refs?service=git-upload-pack" + ) + + def test_open_fetch_logs_ip_only(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + before = len(_log_lines(http_repo_env)) + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + ) + assert resp.status_code == 200 + lines = _log_lines(http_repo_env)[before:] + assert len(lines) == 1 + _assert_log_line( + lines[0], None, project=http_repo_env["project"], operation="info/refs?service=git-upload-pack" + ) + + def test_session_logs_email_without_a_key_tail(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + with _auth_enabled(_user_with_keys()): + token = create_session("alice", email="alice@example.com", auth_method="local") + before = len(_log_lines(http_repo_env)) + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + headers={"Authorization": f"Bearer {token}"}, + ) + assert resp.status_code == 200 + lines = _log_lines(http_repo_env)[before:] + assert len(lines) == 1 + _assert_log_line( + lines[0], + "alice@example.com", + project=http_repo_env["project"], + operation="info/refs?service=git-upload-pack", + ) + + def test_key_owner_without_email_logs_username(self, http_repo_env: dict[str, str]) -> None: + cfg = AuthConfig( + auth_mode="local", + local_users=[ + LocalUser( + user="bob", + password="secret", + keys=[HttpKey(id="bob-key", secret=hash_password("bob-secret"), time_issued=_KEY_ISSUED)], + ) + ], + oauth_permissions=PermissionsMap.default_access(), + ) + client = TestClient(app) + with _auth_enabled(cfg): + before = len(_log_lines(http_repo_env)) + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + auth=("bob", "bob-secret"), + ) + assert resp.status_code == 200 + lines = _log_lines(http_repo_env)[before:] + assert len(lines) == 1 + _assert_log_line( + lines[0], + "bob cret", + project=http_repo_env["project"], + operation="info/refs?service=git-upload-pack", + ) + assert "bob-secret" not in lines[0] + + def test_unknown_project_is_not_logged(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + before = len(_log_lines(http_repo_env)) + resp = client.get("/http/missing/info/refs", params={"service": "git-upload-pack"}) + assert resp.status_code == 404 + assert _log_lines(http_repo_env)[before:] == [] diff --git a/pygitweb/src/pygitweb/static/main.css b/pygitweb/src/pygitweb/static/main.css index 0c42ec9..5fe8547 100644 --- a/pygitweb/src/pygitweb/static/main.css +++ b/pygitweb/src/pygitweb/static/main.css @@ -1033,6 +1033,31 @@ pre { border-left: 0.25rem solid var(--pgw-warn-accent); } +#api-key-secret-panel.alert, +#api-key-alert.alert-success { + --tblr-alert-color: var(--pgw-add-fg); + --tblr-alert-bg: var(--pgw-add-bg); + --tblr-alert-border-color: var(--pgw-border); + background-color: var(--pgw-add-bg); + color: var(--pgw-add-fg); + border-color: var(--pgw-border); +} + +#api-key-alert.alert-danger { + --tblr-alert-color: var(--pgw-rem-fg); + --tblr-alert-bg: var(--pgw-rem-bg); + --tblr-alert-border-color: var(--pgw-border); + background-color: var(--pgw-rem-bg); + color: var(--pgw-rem-fg); + border-color: var(--pgw-border); +} + +#api-key-secret.form-control { + background-color: var(--pgw-bg-raised); + border-color: var(--pgw-border); + color: var(--pgw-text); +} + /* Board: Sync status icons (mask so currentColor is respected when loaded) */ .board-card-icon-wrap { opacity: 0.7; diff --git a/pygitweb/src/pygitweb/static/user-keys.js b/pygitweb/src/pygitweb/static/user-keys.js new file mode 100644 index 0000000..71ced8c --- /dev/null +++ b/pygitweb/src/pygitweb/static/user-keys.js @@ -0,0 +1,109 @@ +const form = document.getElementById("api-key-form"); +const alertBox = document.getElementById("api-key-alert"); +const secretPanel = document.getElementById("api-key-secret-panel"); +const secretInput = document.getElementById("api-key-secret"); +const tableBody = document.querySelector("#api-key-table tbody"); + +function showAlert(message, kind) { + if (!alertBox) { + return; + } + alertBox.textContent = message; + alertBox.className = `alert alert-${kind}`; +} + +function detailMessage(payload) { + if (payload && typeof payload.detail === "string") { + return payload.detail; + } + return "Request failed"; +} + +function appendKeyRow(key) { + const empty = document.getElementById("api-key-empty"); + if (empty) { + empty.remove(); + } + if (!tableBody) { + return; + } + const row = document.createElement("tr"); + row.dataset.keyId = key.id; + const cells = [key.id, key.readonly ? "Read only" : "Read and write", key.time_issued, key.time_expiry || "Never"]; + for (const text of cells) { + const cell = document.createElement("td"); + cell.textContent = text; + row.append(cell); + } + const action = document.createElement("td"); + action.className = "text-end"; + const button = document.createElement("button"); + button.type = "button"; + button.className = "btn btn-outline-danger btn-sm"; + button.dataset.revokeKey = key.id; + button.textContent = "Revoke"; + button.addEventListener("click", () => { + revokeKey(key.id); + }); + action.append(button); + row.append(action); + tableBody.append(row); +} + +async function revokeKey(keyId) { + const response = await fetch(`/user/keys?id=${encodeURIComponent(keyId)}`, { + method: "DELETE", + credentials: "same-origin", + }); + if (!response.ok) { + const payload = await response.json().catch(() => null); + showAlert(detailMessage(payload), "danger"); + return; + } + const row = tableBody?.querySelector(`tr[data-key-id="${CSS.escape(keyId)}"]`); + row?.remove(); + showAlert(`Revoked ${keyId}.`, "success"); + window.location.reload(); +} + +if (form) { + form.addEventListener("submit", async (event) => { + event.preventDefault(); + const idInput = document.getElementById("api-key-id"); + const readonlyInput = document.getElementById("api-key-readonly"); + const keyId = idInput instanceof HTMLInputElement ? idInput.value.trim() : ""; + const readonly = readonlyInput instanceof HTMLInputElement && readonlyInput.checked; + const response = await fetch("/user/keys", { + method: "POST", + credentials: "same-origin", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ id: keyId, readonly }), + }); + const payload = await response.json().catch(() => null); + if (!response.ok) { + showAlert(detailMessage(payload), "danger"); + return; + } + if (secretPanel && secretInput instanceof HTMLInputElement) { + secretInput.value = payload.secret; + secretPanel.classList.remove("d-none"); + } + appendKeyRow(payload); + form.reset(); + showAlert("API key created.", "success"); + const max = Number(document.getElementById("api-key-table")?.dataset.apiKeyMax || "0"); + const held = tableBody ? tableBody.querySelectorAll("tr[data-key-id]").length : 0; + if (max > 0 && held >= max) { + form.remove(); + } + }); +} + +document.querySelectorAll("[data-revoke-key]").forEach((button) => { + button.addEventListener("click", () => { + const keyId = button.getAttribute("data-revoke-key"); + if (keyId) { + revokeKey(keyId); + } + }); +}); diff --git a/pygitweb/src/pygitweb/templates/user_profile.html b/pygitweb/src/pygitweb/templates/user_profile.html index f2f0f90..7031a82 100644 --- a/pygitweb/src/pygitweb/templates/user_profile.html +++ b/pygitweb/src/pygitweb/templates/user_profile.html @@ -1,7 +1,7 @@ {% extends "base.html" %} {% block content %}

Account

-
+

Username

{{ username | e }}

@@ -10,4 +10,76 @@

+ +
+
+

API keys

+ {% if api_key_expiry_days %} +

New keys expire after {{ api_key_expiry_days }} day{% if api_key_expiry_days != 1 %}s{% endif %}. You can hold {{ api_key_max }}.

+ {% else %} +

New keys do not expire. You can hold {{ api_key_max }}.

+ {% endif %} + {% if api_keys is none %} +

No local account matches this session, so API keys cannot be stored.

+ {% else %} + +
+

Copy this key now. It will not be shown again.

+ +
+
+ + + + + + + + + + + + {% for key in api_keys %} + + + + + + + + {% else %} + + {% endfor %} + +
IdAccessIssuedExpires
{{ key.id | e }}{% if key.readonly %}Read only{% else %}Read and write{% endif %}{{ key.time_issued | e }}{% if key.time_expiry %}{{ key.time_expiry | e }}{% else %}Never{% endif %} + +
No API keys yet.
+
+ {% if api_keys | length < api_key_max %} +
+
+ + +
+
+ +
+
+ +
+
+ {% else %} +

Key limit reached. Revoke one to issue another.

+ {% endif %} + {% endif %} +
+
+{% endblock %} +{% block scripts %} +{% if api_keys is not none %} + +{% endif %} {% endblock %}