diff --git a/pygitweb/addproject_permissions_test.py b/pygitweb/addproject_permissions_test.py new file mode 100644 index 0000000..a463ab5 --- /dev/null +++ b/pygitweb/addproject_permissions_test.py @@ -0,0 +1,114 @@ +from collections.abc import Generator +from unittest.mock import patch + +import pytest +from fastapi.testclient import TestClient + +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.main import app +from pygitweb.permissions import PERMISSION_ADD_PROJECTS, PermissionsMap + +_ADDPROJECTS_AUTH = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="admin", password="secret"), LocalUser(user="viewer", password="secret")], + oauth_permissions=PermissionsMap.model_validate({ + "*": [], + PERMISSION_ADD_PROJECTS: ["admin"], + }), +) + + +@pytest.fixture +def client() -> Generator[TestClient, None, None]: + with TestClient(app) as c: + yield c + + +@pytest.fixture +def addprojects_auth() -> Generator[AuthConfig, None, None]: + with ( + patch("pygitweb.auth_config.auth_config", _ADDPROJECTS_AUTH), + patch("pygitweb.auth.auth_config", _ADDPROJECTS_AUTH), + patch("pygitweb.main.auth_config", _ADDPROJECTS_AUTH), + ): + yield _ADDPROJECTS_AUTH + + +def _login(client: TestClient, username: str) -> None: + client.post( + "/login", + data={"username": username, "password": "secret", "next": "/"}, + follow_redirects=False, + ) + + +def test_projectnamevalid_open_when_auth_disabled(client: TestClient) -> None: + from pygitweb.config import settings + + with ( + patch.object(settings, "AUTH", False), + patch("pygitweb.main.project_visible_in_list", return_value=False), + ): + r = client.get("/projectnamevalid", params={"name": "newproj"}) + assert r.status_code == 200 + + +def test_addproject_page_open_when_auth_disabled(client: TestClient) -> None: + from pygitweb.config import settings + + with patch.object(settings, "AUTH", False): + r = client.get("/addproject") + assert r.status_code == 200 + + +def test_projectnamevalid_401_without_login(client: TestClient, addprojects_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with patch.object(settings, "AUTH", True): + r = client.get("/projectnamevalid", params={"name": "newproj"}) + assert r.status_code == 401 + + +def test_projectnamevalid_allowed_with_grant(client: TestClient, addprojects_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with ( + patch.object(settings, "AUTH", True), + patch("pygitweb.main.project_visible_in_list", return_value=False), + ): + _login(client, "admin") + r = client.get("/projectnamevalid", params={"name": "newproj"}) + assert r.status_code == 200 + + +def test_projectnamevalid_forbidden_without_grant(client: TestClient, addprojects_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with ( + patch.object(settings, "AUTH", True), + patch("pygitweb.main.project_visible_in_list", return_value=False), + ): + _login(client, "viewer") + r = client.get("/projectnamevalid", params={"name": "newproj"}) + assert r.status_code == 403 + + +def test_addproject_page_readable_without_grant(client: TestClient, addprojects_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with patch.object(settings, "AUTH", True): + _login(client, "viewer") + r = client.get("/addproject") + assert r.status_code == 200 + + +def test_addproject_post_forbidden_without_grant(client: TestClient, addprojects_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with ( + patch.object(settings, "AUTH", True), + patch("pygitweb.main.project_visible_in_list", return_value=False), + ): + _login(client, "viewer") + r = client.post("/addproject", data={"project_name": "newproj"}, follow_redirects=False) + assert r.status_code == 403 diff --git a/pygitweb/auth.py b/pygitweb/auth.py index cb63480..71b28fd 100644 --- a/pygitweb/auth.py +++ b/pygitweb/auth.py @@ -30,6 +30,7 @@ from pygitweb.auth_config import ( ) from pygitweb.auth_oauth import OAUTH_STATE_COOKIE_NAME, oauth_router from pygitweb.config import settings +from pygitweb.permissions import Permission, PermissionPrincipal from pygitweb.sessions import create_session, get_session, revoke_session from pygitweb.templates_env import POSTAMBLE, PREAMBLE, env @@ -49,6 +50,81 @@ auth_router.include_router(oauth_router) class User(BaseModel): username: str + email: str | None = None + + +def principal_from_session(token: str | None) -> PermissionPrincipal | None: + if not token: + return None + rec = get_session(token) + if rec is None: + return None + return PermissionPrincipal(username=rec.username, email=rec.email) + + +def has_permission( + permission: Permission, + project: str | None = None, + *, + token: str | None = None, +) -> bool: + if not settings.AUTH: + return True + principal = principal_from_session(token) + if principal is None: + return False + return auth_config.oauth_permissions.has_permission(principal.identity, permission, project) + + +def ensure_permission( + permission: Permission, + project: str | None = None, + *, + token: str | None, +) -> None: + ensure_active_user_if_auth_enabled(token) + if has_permission(permission, project, token=token): + return + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Insufficient permissions", + ) + + +def require_permission( + permission: Permission, + *, + project_from_query: bool = False, + project_from_form: bool = False, +) -> object: + """Return a FastAPI dependency that enforces ``permission`` when auth is enabled.""" + + if project_from_query: + + def _dep_query_project( + token: Annotated[str | None, Depends(access_token_from_request)], + project: Annotated[str, Query()], + ) -> None: + ensure_permission(permission, project, token=token) + + return _dep_query_project + + if project_from_form: + + def _dep_form_project( + token: Annotated[str | None, Depends(access_token_from_request)], + project: Annotated[str, Form()], + ) -> None: + ensure_permission(permission, project, token=token) + + return _dep_form_project + + def _dep( + token: Annotated[str | None, Depends(access_token_from_request)], + ) -> None: + ensure_permission(permission, None, token=token) + + return _dep class UserInDB(User): diff --git a/pygitweb/auth.schema.json b/pygitweb/auth.schema.json index f704742..ee968e1 100644 --- a/pygitweb/auth.schema.json +++ b/pygitweb/auth.schema.json @@ -36,18 +36,26 @@ }, "oauth_permissions": { "type": "object", + "description": "Map of permission keys to principal emails (or local usernames). Key * grants all permissions.", "properties": { "*": { "type": "array", - "items": { - "required": [], - "properties": {} - } + "items": { "type": "string" } + }, + "pgw.addprojects": { + "type": "array", + "items": { "type": "string" } + }, + "pgw.settings": { + "type": "array", + "items": { "type": "string" } } }, - "required": [ - "*" - ] + "additionalProperties": { + "type": "array", + "items": { "type": "string" }, + "description": "Scoped keys: pgw.boards.{scope}, pgw.tasks.{scope}, pgw.comments.{scope}, pgw.hooks.{scope}, pgw.mr.create.{scope}, pgw.mr.merge.{scope} (fnmatch on project path)" + } }, "local_users": { "type": "array", diff --git a/pygitweb/auth_config.py b/pygitweb/auth_config.py index dcab494..b30d979 100644 --- a/pygitweb/auth_config.py +++ b/pygitweb/auth_config.py @@ -14,9 +14,10 @@ import sys from pathlib import Path from typing import Literal -from pydantic import BaseModel, ConfigDict, Field, field_validator +from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator from pygitweb.config import Settings +from pygitweb.permissions import PermissionsMap DEFAULT_AUTH_CONFIG_PATH = Path.home() / ".pygitweb" / "auth.json" _AUTH_CONFIG_MODE = 0o600 @@ -72,17 +73,19 @@ class AuthConfig(BaseModel): oauth_redirect_uri: str = "" oauth_username_from: OAuthUsernameFrom = "email" oauth_allowed_emails: list[str] = Field(default_factory=list) - oauth_permissions: dict[str, list[str]] = Field(default_factory=lambda: {"*": []}) + oauth_permissions: PermissionsMap = Field(default_factory=PermissionsMap.empty) local_users: list[LocalUser] = Field(default_factory=list) @field_validator("oauth_permissions", mode="before") @classmethod - def _ensure_wildcard_permissions(cls, v: object) -> object: - if v is None: - return {"*": []} - if isinstance(v, dict) and "*" not in v: - return {**v, "*": []} - return v + def _validate_oauth_permissions(cls, v: object) -> PermissionsMap | dict[str, list[str]]: + if isinstance(v, PermissionsMap): + return v + return PermissionsMap.model_validate(v) + + @field_serializer("oauth_permissions") + def _serialize_oauth_permissions(self, value: PermissionsMap) -> dict[str, list[str]]: + return value.root def auth_config_path(settings: Settings) -> Path: @@ -100,7 +103,7 @@ def default_auth_config(*, admin_password: str) -> AuthConfig: oauth_redirect_uri="", oauth_username_from="email", oauth_allowed_emails=[], - oauth_permissions={"*": []}, + oauth_permissions=PermissionsMap.empty(), local_users=[LocalUser(user="admin", password=admin_password)], ) diff --git a/pygitweb/auth_test.py b/pygitweb/auth_test.py index 35fc19b..3eaae25 100644 --- a/pygitweb/auth_test.py +++ b/pygitweb/auth_test.py @@ -10,7 +10,9 @@ from fastapi.testclient import TestClient from pygitweb.auth_config import AuthConfig, LocalUser from pygitweb.auth_oauth import OAUTH_STATE_COOKIE_NAME, _seal_oauth_state from pygitweb.config import settings +from pygitweb.conftest import clear_client_cookies from pygitweb.main import app +from pygitweb.permissions import PERMISSION_ADD_PROJECTS, PermissionsMap from pygitweb.sessions import get_session _LOCAL_ADMIN = AuthConfig( @@ -118,9 +120,20 @@ def test_projectnamevalid_401_without_credentials_when_auth_enabled( def test_projectnamevalid_ok_with_bearer_when_auth_enabled(client: TestClient, local_auth_config: AuthConfig) -> None: + cfg = local_auth_config.model_copy( + update={ + "oauth_permissions": PermissionsMap.model_validate({ + "*": [], + PERMISSION_ADD_PROJECTS: ["admin"], + }), + }, + ) with ( patch.object(settings, "AUTH", True), patch("pygitweb.main.project_visible_in_list", return_value=False), + patch("pygitweb.auth_config.auth_config", cfg), + patch("pygitweb.auth.auth_config", cfg), + patch("pygitweb.main.auth_config", cfg), ): tok = client.post("/token", data={"username": "admin", "password": "secret"}).json()["access_token"] r = client.get( @@ -244,10 +257,10 @@ def test_oauth_callback_creates_session(client: TestClient, oauth_auth_config: A patch("pygitweb.auth_oauth._exchange_code", new_callable=AsyncMock, return_value=token_payload), patch("pygitweb.auth_oauth._fetch_profile", new_callable=AsyncMock, return_value=profile), ): + client.cookies.set(OAUTH_STATE_COOKIE_NAME, sealed) r = client.get( "/auth/oauth/callback", params={"code": "auth-code", "state": "idp-state"}, - cookies={OAUTH_STATE_COOKIE_NAME: sealed}, follow_redirects=False, ) assert r.status_code == 303 @@ -258,6 +271,7 @@ def test_oauth_callback_creates_session(client: TestClient, oauth_auth_config: A assert rec.email == "user@example.com" assert rec.auth_method == "oauth" assert client.get("/users/me").json()["username"] == "user@example.com" + clear_client_cookies(client) def test_authenticate_local_user_disabled_in_oauth_only_mode() -> None: diff --git a/pygitweb/boards_permissions_test.py b/pygitweb/boards_permissions_test.py new file mode 100644 index 0000000..399d5d5 --- /dev/null +++ b/pygitweb/boards_permissions_test.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +from collections.abc import Generator +from pathlib import Path +from unittest.mock import patch + +import pygit2 +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.config import settings +from pygitweb.conftest import client_as_user +from pygitweb.permissions import Permission, PermissionsMap, permission_key +from pygitweb.tasks import board_router + +_BOARDS_AUTH = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="admin", password="secret"), LocalUser(user="viewer", password="secret")], + oauth_permissions=PermissionsMap.model_validate({ + "*": [], + permission_key(Permission.BOARDS, "demo"): ["admin"], + }), +) + + +def _build_client() -> TestClient: + app = FastAPI() + app.include_router(board_router, prefix="/board") + return TestClient(app) + + +def _create_initial_commit(repo: pygit2.Repository, repo_dir: Path) -> None: + (repo_dir / "README.md").write_text("seed\n", encoding="utf-8") + index = repo.index + index.add("README.md") + index.write() + tree = index.write_tree() + sig = pygit2.Signature("tester", "tester@example.com") + repo.create_commit("HEAD", sig, sig, "initial", tree, []) + + +@pytest.fixture +def board_env(tmp_path: Path) -> Generator[dict[str, str | TestClient], None, None]: + repo_dir = tmp_path / "demo" + repo = pygit2.init_repository(str(repo_dir), bare=False) + _create_initial_commit(repo, repo_dir) + with ( + patch.object(settings, "PROJECTROOT", str(tmp_path)), + patch.object(settings, "PROJECTS_LIST", str(tmp_path)), + patch.object(settings, "STRICT_EXPORT", False), + patch.object(settings, "EXPORT_OK", ""), + patch.object(settings, "LIST_ALL", True), + patch("pygitweb.auth_config.auth_config", _BOARDS_AUTH), + patch("pygitweb.auth.auth_config", _BOARDS_AUTH), + ): + yield {"client": _build_client(), "project": "demo"} + + +def test_board_post_open_when_auth_disabled(board_env: dict[str, str | TestClient]) -> None: + client = board_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", False): + r = client.post( + "/board/create", + params={"project": board_env["project"], "name": "temp"}, + ) + assert r.status_code == 200 + + +def test_board_create_401_without_login(board_env: dict[str, str | TestClient]) -> None: + client = board_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True): + r = client.post( + "/board/create", + params={"project": board_env["project"], "name": "temp"}, + ) + assert r.status_code == 401 + + +def test_board_create_allowed_with_grant(board_env: dict[str, str | TestClient]) -> None: + client = board_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "admin"): + r = client.post( + "/board/create", + params={"project": board_env["project"], "name": "temp-board"}, + ) + assert r.status_code == 200 + + +def test_board_create_forbidden_without_grant(board_env: dict[str, str | TestClient]) -> None: + client = board_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "viewer"): + r = client.post( + "/board/create", + params={"project": board_env["project"], "name": "temp-board"}, + ) + assert r.status_code == 403 + + +def test_board_delete_forbidden_without_grant(board_env: dict[str, str | TestClient]) -> None: + client = board_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "viewer"): + r = client.post( + "/board/delete", + params={"project": board_env["project"], "name": "Tasks"}, + ) + assert r.status_code == 403 diff --git a/pygitweb/comments_permissions_test.py b/pygitweb/comments_permissions_test.py new file mode 100644 index 0000000..39682bd --- /dev/null +++ b/pygitweb/comments_permissions_test.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +from collections.abc import Generator +from pathlib import Path +from unittest.mock import patch + +import pygit2 +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from pygittools.tasks import BOARD_REF_PREFIX, TASK_REF_PREFIX, Board, Task +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.config import settings +from pygitweb.conftest import client_as_user +from pygitweb.permissions import Permission, PermissionsMap, permission_key +from pygitweb.tasks import EMPTY_TREE_OID, comment_router + +_COMMENTS_AUTH = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="admin", password="secret"), LocalUser(user="viewer", password="secret")], + oauth_permissions=PermissionsMap.model_validate({ + "*": [], + permission_key(Permission.COMMENTS, "demo"): ["viewer"], + permission_key(Permission.TASKS, "demo"): ["admin"], + }), +) + + +def _build_client() -> TestClient: + app = FastAPI() + app.include_router(comment_router, prefix="/comments") + return TestClient(app) + + +def _create_initial_commit(repo: pygit2.Repository, repo_dir: Path) -> None: + (repo_dir / "README.md").write_text("seed\n", encoding="utf-8") + index = repo.index + index.add("README.md") + index.write() + tree = index.write_tree() + sig = pygit2.Signature("tester", "tester@example.com") + repo.create_commit("HEAD", sig, sig, "initial", tree, []) + + +@pytest.fixture +def comment_env(tmp_path: Path) -> Generator[dict[str, str], None, None]: + repo_dir = tmp_path / "demo" + repo = pygit2.init_repository(str(repo_dir), bare=False) + _create_initial_commit(repo, repo_dir) + board_ref = f"{BOARD_REF_PREFIX}Tasks" + board = Board(EMPTY_TREE_OID, board_ref, tagger="", description="") + task_ref = f"{TASK_REF_PREFIX}task_1" + task = Task(EMPTY_TREE_OID, task_ref, tagger="", title="Task", description="") + task_oid = task.write(repo) + board.tasks = [str(task_oid)] + board.update_message() + board.write(repo) + with ( + patch.object(settings, "PROJECTROOT", str(tmp_path)), + patch.object(settings, "PROJECTS_LIST", str(tmp_path)), + patch.object(settings, "STRICT_EXPORT", False), + patch.object(settings, "EXPORT_OK", ""), + patch.object(settings, "LIST_ALL", True), + patch("pygitweb.auth_config.auth_config", _COMMENTS_AUTH), + patch("pygitweb.auth.auth_config", _COMMENTS_AUTH), + ): + client = _build_client() + yield { + "client": client, + "project": "demo", + "board": "Tasks", + "task": task_ref, + } + + +def test_comment_create_open_when_auth_disabled(comment_env: dict[str, str]) -> None: + client = comment_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", False): + r = client.post( + "/comments/create", + params={ + "project": comment_env["project"], + "board": comment_env["board"], + "task": comment_env["task"], + "content": "note", + }, + ) + assert r.status_code == 200 + + +def test_comment_create_401_without_login(comment_env: dict[str, str]) -> None: + client = comment_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True): + r = client.post( + "/comments/create", + params={ + "project": comment_env["project"], + "board": comment_env["board"], + "task": comment_env["task"], + "content": "note", + }, + ) + assert r.status_code == 401 + + +def test_comment_create_allowed_with_comments_grant_only(comment_env: dict[str, str]) -> None: + """User with COMMENTS but not TASKS can add a comment.""" + client = comment_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "viewer"): + r = client.post( + "/comments/create", + params={ + "project": comment_env["project"], + "board": comment_env["board"], + "task": comment_env["task"], + "content": "from viewer", + }, + ) + assert r.status_code == 200 + + +def test_comment_create_forbidden_without_comments_grant(comment_env: dict[str, str]) -> None: + """User with TASKS but not COMMENTS cannot add a comment.""" + client = comment_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "admin"): + r = client.post( + "/comments/create", + params={ + "project": comment_env["project"], + "board": comment_env["board"], + "task": comment_env["task"], + "content": "from admin", + }, + ) + assert r.status_code == 403 diff --git a/pygitweb/conftest.py b/pygitweb/conftest.py index 71c6ab2..07fc468 100644 --- a/pygitweb/conftest.py +++ b/pygitweb/conftest.py @@ -4,15 +4,38 @@ from __future__ import annotations import asyncio import os +from collections.abc import Generator +from contextlib import contextmanager from pathlib import Path import pytest +from fastapi.testclient import TestClient _test_auth_config = Path(os.environ.get("TEMP", "/tmp")) / "pygitweb-test-auth.json" os.environ.setdefault("PYGITWEB_AUTH", "0") os.environ.setdefault("PYGITWEB_AUTH_CONFIG", str(_test_auth_config)) +def clear_client_cookies(client: TestClient) -> None: + client.cookies.clear() + + +def set_client_session(client: TestClient, username: str) -> None: + from pygitweb.auth import ACCESS_TOKEN_COOKIE_NAME + from pygitweb.sessions import create_session + + client.cookies.set(ACCESS_TOKEN_COOKIE_NAME, create_session(username, auth_method="local")) + + +@contextmanager +def client_as_user(client: TestClient, username: str) -> Generator[None, None, None]: + set_client_session(client, username) + try: + yield + finally: + clear_client_cookies(client) + + @pytest.fixture(autouse=True) def _reset_pygitweb_app_lifecycle_state() -> None: """Starlette TestClient lifespan sets shutting_down on exit; httpx ASGITransport never runs lifespan. diff --git a/pygitweb/hooks_permissions_test.py b/pygitweb/hooks_permissions_test.py new file mode 100644 index 0000000..55ca8e5 --- /dev/null +++ b/pygitweb/hooks_permissions_test.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from collections.abc import Generator +from pathlib import Path +from unittest.mock import patch + +import pygit2 +import pytest +from fastapi.testclient import TestClient + +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.config import settings +from pygitweb.conftest import client_as_user +from pygitweb.main import app +from pygitweb.permissions import Permission, PermissionsMap, permission_key + +_HOOKS_AUTH = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="admin", password="secret"), LocalUser(user="viewer", password="secret")], + oauth_permissions=PermissionsMap.model_validate({ + "*": [], + permission_key(Permission.HOOKS, "demo"): ["admin"], + }), +) + + +def _create_initial_commit(repo: pygit2.Repository, repo_dir: Path) -> None: + (repo_dir / "README.md").write_text("seed\n", encoding="utf-8") + index = repo.index + index.add("README.md") + index.write() + tree = index.write_tree() + sig = pygit2.Signature("tester", "tester@example.com") + repo.create_commit("HEAD", sig, sig, "initial", tree, []) + + +@pytest.fixture +def hook_env(tmp_path: Path) -> Generator[dict[str, str], None, None]: + root = tmp_path / "root" + root.mkdir() + repo_dir = root / "demo" + repo_dir.mkdir() + _create_initial_commit(pygit2.init_repository(str(repo_dir), bare=False), repo_dir) + with ( + patch.object(settings, "PROJECTROOT", str(root)), + patch.object(settings, "PROJECTS_LIST", str(root)), + patch.object(settings, "PROJECT_MAXDEPTH", 3), + patch.object(settings, "STRICT_EXPORT", False), + patch.object(settings, "EXPORT_OK", ""), + patch.object(settings, "LIST_ALL", True), + patch.object(settings, "MAXLOAD", None), + patch("pygitweb.auth_config.auth_config", _HOOKS_AUTH), + patch("pygitweb.auth.auth_config", _HOOKS_AUTH), + patch("pygitweb.main.auth_config", _HOOKS_AUTH), + ): + yield {"project": "demo"} + + +@pytest.fixture +def client() -> Generator[TestClient, None, None]: + with TestClient(app) as c: + yield c + + +def test_hook_check_open_when_auth_disabled(client: TestClient, hook_env: dict[str, str]) -> None: + with patch.object(settings, "AUTH", False): + r = client.post( + f"/project/{hook_env['project']}/hook", + params={"name": "post-receive.notify", "op": "check"}, + ) + assert r.status_code == 200 + + +def test_hook_add_open_when_auth_disabled(client: TestClient, hook_env: dict[str, str]) -> None: + with patch.object(settings, "AUTH", False): + r = client.post( + f"/project/{hook_env['project']}/hook", + params={"name": "post-receive.notify", "op": "add"}, + ) + assert r.status_code == 200 + + +def test_hook_add_401_without_login(client: TestClient, hook_env: dict[str, str]) -> None: + with patch.object(settings, "AUTH", True): + r = client.post( + f"/project/{hook_env['project']}/hook", + params={"name": "post-receive.notify", "op": "add"}, + ) + assert r.status_code == 401 + + +def test_hook_add_allowed_with_grant(client: TestClient, hook_env: dict[str, str]) -> None: + with patch.object(settings, "AUTH", True), client_as_user(client, "admin"): + r = client.post( + f"/project/{hook_env['project']}/hook", + params={"name": "post-receive.notify", "op": "add"}, + ) + assert r.status_code == 200 + + +def test_hook_add_forbidden_without_grant(client: TestClient, hook_env: dict[str, str]) -> None: + with patch.object(settings, "AUTH", True), client_as_user(client, "viewer"): + r = client.post( + f"/project/{hook_env['project']}/hook", + params={"name": "post-receive.notify", "op": "add"}, + ) + assert r.status_code == 403 + + +def test_hook_check_allowed_without_hooks_grant(client: TestClient, hook_env: dict[str, str]) -> None: + with patch.object(settings, "AUTH", True), client_as_user(client, "viewer"): + r = client.post( + f"/project/{hook_env['project']}/hook", + params={"name": "post-receive.notify", "op": "check"}, + ) + assert r.status_code == 200 diff --git a/pygitweb/main.py b/pygitweb/main.py index 0335d87..16f15a5 100644 --- a/pygitweb/main.py +++ b/pygitweb/main.py @@ -58,8 +58,9 @@ from pygitweb.auth import ( access_token_from_request, auth_router, decode_access_token, - ensure_active_user_if_auth_enabled, + ensure_permission, require_active_user_if_auth_enabled, + require_permission, ) from pygitweb.auth_config import auth_config, is_auth_configured from pygitweb.change_queue import CHANGE_QUEUE @@ -96,6 +97,7 @@ from pygitweb.hooks_install import ( status as hook_status, ) from pygitweb.merge_requests import merge_router +from pygitweb.permissions import Permission from pygitweb.plugin_loader import load_plugin_actions, load_subpages from pygitweb.projects import git_get_project_owner, git_get_projects_list from pygitweb.sessions import clear_all_sessions @@ -448,10 +450,10 @@ _OPTIONAL_REPO_ZIP = File(default=None) @app.get("/projectnamevalid", response_class=HTMLResponse) def addproject_namevalid( - _: Annotated[None, Depends(require_active_user_if_auth_enabled)], + _: Annotated[None, Depends(require_permission(Permission.ADD_PROJECTS))], name: Annotated[str | None, Query()] = None, ): - """Check if project name is valid (authenticated when PYGITWEB auth is enabled).""" + """Check if project name is valid (requires pgw.addprojects when auth is enabled).""" if not name: raise HTTPException(status_code=400, detail="Param 'name' required") if not is_valid_pathname(name): @@ -485,7 +487,7 @@ def addproject_page( @app.post("/addproject", response_class=HTMLResponse) async def addproject_submit( - _: Annotated[None, Depends(require_active_user_if_auth_enabled)], + _: Annotated[None, Depends(require_permission(Permission.ADD_PROJECTS))], request: Request, project_name: Annotated[str, Form()] = "", pull_from_remote: Annotated[str, Form()] = "", @@ -493,10 +495,7 @@ async def addproject_submit( create_task_board: Annotated[str, Form()] = "off", repo_zip: UploadFile | None = _OPTIONAL_REPO_ZIP, ): - """ - Create a new project. Allowed only if project name does not exist, - and a valid Bearer token when authentication is enabled. - """ + """Create a new project (requires pgw.addprojects when auth is enabled).""" project_name = (project_name or "").strip() if not project_name: raise HTTPException(status_code=400, detail="Project name is required") @@ -640,7 +639,7 @@ def project_hook( `op` is `add`, `remove`, or `check`. """ if op != "check": - ensure_active_user_if_auth_enabled(token) + ensure_permission(Permission.HOOKS, project, token=token) bundle = get_bundle(name) sample = get_sample(name) if bundle is None else None if bundle is None and sample is None: diff --git a/pygitweb/merge_requests.py b/pygitweb/merge_requests.py index 8c0d761..77c6863 100644 --- a/pygitweb/merge_requests.py +++ b/pygitweb/merge_requests.py @@ -11,10 +11,11 @@ from fastapi import APIRouter, Depends, Form, HTTPException, Query from fastapi.responses import HTMLResponse, RedirectResponse, Response from pygittools.merge import MergeRequest, MergeRequestStatus, get_merge_request_by_oid -from pygitweb.auth import require_active_user_if_auth_enabled +from pygitweb.auth import require_permission from pygitweb.config import settings from pygitweb.dependencies import ValidatedFormProject, ValidatedQueryProject from pygitweb.git_helpers import open_repo +from pygitweb.permissions import Permission from pygitweb.templates_env import POSTAMBLE, PREAMBLE, env, jinja_escape from pygitweb.validation import is_valid_ref_format @@ -23,7 +24,7 @@ def _branch_short_name(ref: str) -> str: return ref.removeprefix("refs/heads/") if ref.startswith("refs/heads/") else ref -merge_router = APIRouter(tags=["merge_requests"], dependencies=[Depends(require_active_user_if_auth_enabled)]) +merge_router = APIRouter(tags=["merge_requests"]) _STATUS_LABELS: dict[MergeRequestStatus, str] = { MergeRequestStatus.CAN_FAST_FORWARD: "Can fast-forward", @@ -156,7 +157,10 @@ def merge_request_tag_response( return HTMLResponse(f"{pre}{body}{POSTAMBLE}") -@merge_router.post("/create") +@merge_router.post( + "/create", + dependencies=[Depends(require_permission(Permission.MR_CREATE, project_from_form=True))], +) def merge_request_create( project: ValidatedFormProject, theirs: str = Form(...), @@ -195,7 +199,10 @@ def merge_request_create( return RedirectResponse(url=dest, status_code=303) -@merge_router.post("/ff") +@merge_router.post( + "/ff", + dependencies=[Depends(require_permission(Permission.MR_MERGE, project_from_query=True))], +) def merge_fast_forward( project: ValidatedQueryProject, h: str = Query(..., description="Annotated tag object id (hex)"), diff --git a/pygitweb/mr_permissions_test.py b/pygitweb/mr_permissions_test.py new file mode 100644 index 0000000..62f726d --- /dev/null +++ b/pygitweb/mr_permissions_test.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +from collections.abc import Generator +from pathlib import Path +from unittest.mock import patch + +import pygit2 +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from pygittools.merge import MR_REF_PREFIX, MergeRequest +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.config import settings +from pygitweb.conftest import client_as_user +from pygitweb.merge_requests import merge_router +from pygitweb.permissions import Permission, PermissionsMap, permission_key + + +def _commit_chain(repo: pygit2.Repository) -> tuple[pygit2.Oid, pygit2.Oid]: + sig = pygit2.Signature("tester", "tester@example.com") + tb = repo.TreeBuilder() + tree = tb.write() + a = repo.create_commit(None, sig, sig, "a", tree, []) + repo.create_reference("refs/heads/main", a) + b = repo.create_commit(None, sig, sig, "b", tree, [a]) + return a, b + + +_MR_AUTH = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="creator", password="secret"), LocalUser(user="merger", password="secret")], + oauth_permissions=PermissionsMap.model_validate({ + "*": [], + permission_key(Permission.MR_CREATE, "demo"): ["creator"], + permission_key(Permission.MR_MERGE, "demo"): ["merger"], + }), +) + + +def _build_client() -> TestClient: + app = FastAPI() + app.include_router(merge_router, prefix="/mr") + return TestClient(app) + + +@pytest.fixture +def mr_env(tmp_path: Path) -> Generator[dict[str, str], None, None]: + root = tmp_path / "mr-perms" + root.mkdir() + repo_dir = root / "demo" + repo_dir.mkdir() + repo = pygit2.init_repository(str(repo_dir), bare=False) + _a, b = _commit_chain(repo) + mr = MergeRequest( + b, + repo.default_signature, + ours="refs/heads/main", + title="Feature MR", + name=f"{MR_REF_PREFIX}aaaabbbbccccdddd", + ) + mr_oid = mr.write(repo) + projects_list = root / "projects.list" + projects_list.write_text("demo tester\n", encoding="utf-8") + with ( + patch.object(settings, "PROJECTROOT", str(root)), + patch.object(settings, "PROJECTS_LIST", str(projects_list)), + patch.object(settings, "STRICT_EXPORT", False), + patch.object(settings, "EXPORT_OK", ""), + patch("pygitweb.auth_config.auth_config", _MR_AUTH), + patch("pygitweb.auth.auth_config", _MR_AUTH), + ): + yield { + "client": _build_client(), + "project": "demo", + "mr_tag_oid": str(mr_oid), + "tip_b": str(b), + } + + +def test_mr_ff_open_when_auth_disabled(mr_env: dict[str, str]) -> None: + client = mr_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", False): + r = client.post( + "/mr/ff", + params={"project": mr_env["project"], "h": mr_env["mr_tag_oid"]}, + follow_redirects=False, + ) + assert r.status_code == 303 + + +def test_mr_create_401_without_login(mr_env: dict[str, str]) -> None: + client = mr_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True): + r = client.post( + "/mr/create", + data={ + "project": mr_env["project"], + "theirs": "refs/heads/main", + "ours": "refs/heads/main", + "title": "MR", + }, + follow_redirects=False, + ) + assert r.status_code == 401 + + +def test_mr_create_forbidden_without_create_grant(mr_env: dict[str, str]) -> None: + client = mr_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "merger"): + r = client.post( + "/mr/create", + data={ + "project": mr_env["project"], + "theirs": "refs/heads/main", + "ours": "refs/heads/main", + "title": "MR", + }, + follow_redirects=False, + ) + assert r.status_code == 403 + + +def test_mr_ff_forbidden_without_merge_grant(mr_env: dict[str, str]) -> None: + client = mr_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "creator"): + r = client.post( + "/mr/ff", + params={"project": mr_env["project"], "h": mr_env["mr_tag_oid"]}, + follow_redirects=False, + ) + assert r.status_code == 403 + + +def test_mr_ff_allowed_with_merge_grant(mr_env: dict[str, str]) -> None: + client = mr_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "merger"): + r = client.post( + "/mr/ff", + params={"project": mr_env["project"], "h": mr_env["mr_tag_oid"]}, + follow_redirects=False, + ) + assert r.status_code == 303 diff --git a/pygitweb/permissions.py b/pygitweb/permissions.py new file mode 100644 index 0000000..125f79d --- /dev/null +++ b/pygitweb/permissions.py @@ -0,0 +1,209 @@ +""" +PyGitWeb permission keys and grant checks. + +``auth.json`` field ``oauth_permissions`` maps permission keys to principal lists +(typically OAuth email addresses). The special key ``*`` lists principals granted +every permission. + +Scoped keys use a project segment that supports ``fnmatch`` globs (``*``, ``?``, +``[seq]``), e.g. ``pgw.tasks.foo/*`` or ``pgw.boards.*``. +""" + +from __future__ import annotations + +import fnmatch +import re +from enum import StrEnum +from typing import Self + +from pydantic import BaseModel, ConfigDict, RootModel, model_validator + +WILDCARD_PERMISSION_KEY = "*" + +# Known permission families (suffix after ``pgw.``). +PERMISSION_ADD_PROJECTS = "pgw.addprojects" +PERMISSION_SETTINGS = "pgw.settings" + +_SCOPED_PREFIX = re.compile(r"^pgw\.(boards|tasks|comments|hooks|mr\.create|mr\.merge)\.(.+)$") + + +class Permission(StrEnum): + """Logical permission checked by route dependencies.""" + + BOARDS = "boards" + TASKS = "tasks" + COMMENTS = "comments" + HOOKS = "hooks" + MR_CREATE = "mr.create" + MR_MERGE = "mr.merge" + ADD_PROJECTS = "addprojects" + SETTINGS = "settings" + + @property + def global_key(self) -> str | None: + if self is Permission.ADD_PROJECTS: + return PERMISSION_ADD_PROJECTS + if self is Permission.SETTINGS: + return PERMISSION_SETTINGS + return None + + @property + def scoped_prefix(self) -> str | None: + if self in ( + Permission.BOARDS, + Permission.TASKS, + Permission.COMMENTS, + Permission.HOOKS, + Permission.MR_CREATE, + Permission.MR_MERGE, + ): + return f"pgw.{self.value}." + return None + + +class ParsedPermissionKey(BaseModel): + model_config = ConfigDict(frozen=True) + + raw: str + permission: Permission + project_scope: str | None = None + + @property + def is_global(self) -> bool: + return self.project_scope is None + + +def parse_permission_key(key: str) -> ParsedPermissionKey | None: + """Return a parsed key, or ``None`` if ``key`` is not a known permission name.""" + raw = key.strip() + if not raw or raw == WILDCARD_PERMISSION_KEY: + return None + if raw == PERMISSION_ADD_PROJECTS: + return ParsedPermissionKey(raw=raw, permission=Permission.ADD_PROJECTS) + if raw == PERMISSION_SETTINGS: + return ParsedPermissionKey(raw=raw, permission=Permission.SETTINGS) + m = _SCOPED_PREFIX.match(raw) + if not m: + return None + kind, scope = m.group(1), m.group(2) + try: + perm = Permission(kind) + except ValueError: + return None + if not scope: + return None + return ParsedPermissionKey(raw=raw, permission=perm, project_scope=scope) + + +def permission_key(permission: Permission, project: str) -> str: + """Build the canonical scoped permission key for a project.""" + prefix = permission.scoped_prefix + if prefix is None: + raise ValueError(f"{permission!r} is not project-scoped") + return f"{prefix}{project}" + + +def project_matches_scope(project: str, scope: str) -> bool: + if scope == WILDCARD_PERMISSION_KEY: + return True + return fnmatch.fnmatchcase(project, scope) + + +class PermissionsMap(RootModel[dict[str, list[str]]]): + """Validated ``oauth_permissions`` object from auth.json.""" + + root: dict[str, list[str]] + + @model_validator(mode="before") + @classmethod + def _coerce_root(cls, v: object) -> object: + if isinstance(v, PermissionsMap): + return v.root + if v is None: + return {WILDCARD_PERMISSION_KEY: []} + if not isinstance(v, dict): + raise TypeError("oauth_permissions must be an object") + out: dict[str, list[str]] = {} + for key, principals in v.items(): + if not isinstance(key, str): + raise TypeError("permission keys must be strings") + if key != WILDCARD_PERMISSION_KEY and parse_permission_key(key) is None: + raise ValueError(f"unknown permission key: {key!r}") + if not isinstance(principals, list): + raise TypeError(f"oauth_permissions[{key!r}] must be a list") + normalized: list[str] = [] + for p in principals: + if not isinstance(p, str): + raise TypeError(f"oauth_permissions[{key!r}] entries must be strings") + s = p.strip() + if s: + normalized.append(s) + out[key] = normalized + if WILDCARD_PERMISSION_KEY not in out: + out = {**out, WILDCARD_PERMISSION_KEY: []} + return out + + @classmethod + def empty(cls) -> Self: + return cls({WILDCARD_PERMISSION_KEY: []}) + + def principals_for_key(self, key: str) -> frozenset[str]: + return frozenset(self.root.get(key, [])) + + def superuser_principals(self) -> frozenset[str]: + return self.principals_for_key(WILDCARD_PERMISSION_KEY) + + def matching_scoped_principals(self, permission: Permission, project: str) -> frozenset[str]: + prefix = permission.scoped_prefix + if prefix is None: + return frozenset() + acc: set[str] = set() + for key, principals in self.root.items(): + if key == WILDCARD_PERMISSION_KEY: + continue + parsed = parse_permission_key(key) + if parsed is None or parsed.permission is not permission or parsed.project_scope is None: + continue + if project_matches_scope(project, parsed.project_scope): + acc.update(principals) + return frozenset(acc) + + def principals_with_permission(self, permission: Permission, project: str | None = None) -> frozenset[str]: + acc = set(self.superuser_principals()) + global_key = permission.global_key + if global_key is not None: + acc.update(self.principals_for_key(global_key)) + return frozenset(acc) + if project is None: + return frozenset(acc) + acc.update(self.matching_scoped_principals(permission, project)) + return frozenset(acc) + + def has_permission(self, principal: str | None, permission: Permission, project: str | None = None) -> bool: + if not principal or not principal.strip(): + return False + normalized = principal.strip() + allowed = self.principals_with_permission(permission, project) + if normalized in allowed: + return True + # Case-insensitive match for email principals. + lower = normalized.lower() + return any(p.lower() == lower for p in allowed) + + +def normalize_oauth_permissions(raw: object) -> dict[str, list[str]]: + return PermissionsMap.model_validate(raw).root + + +class PermissionPrincipal(BaseModel): + model_config = ConfigDict(frozen=True) + + username: str + email: str | None = None + + @property + def identity(self) -> str: + """Principal string used for grant lookup (email when present, else username).""" + if self.email and self.email.strip(): + return self.email.strip() + return self.username.strip() diff --git a/pygitweb/permissions_test.py b/pygitweb/permissions_test.py new file mode 100644 index 0000000..ff01d31 --- /dev/null +++ b/pygitweb/permissions_test.py @@ -0,0 +1,86 @@ +import pytest + +from pygitweb.permissions import ( + PERMISSION_ADD_PROJECTS, + PERMISSION_SETTINGS, + Permission, + PermissionsMap, + parse_permission_key, + permission_key, + project_matches_scope, +) + + +def test_parse_global_keys() -> None: + assert parse_permission_key(PERMISSION_ADD_PROJECTS) is not None + assert parse_permission_key(PERMISSION_SETTINGS) is not None + assert parse_permission_key("pgw.unknown") is None + assert parse_permission_key("*") is None + + +def test_parse_scoped_key() -> None: + parsed = parse_permission_key("pgw.tasks.my/repo") + assert parsed is not None + assert parsed.permission is Permission.TASKS + assert parsed.project_scope == "my/repo" + + +def test_permission_key_roundtrip() -> None: + assert permission_key(Permission.BOARDS, "foo") == "pgw.boards.foo" + assert permission_key(Permission.MR_CREATE, "foo") == "pgw.mr.create.foo" + assert permission_key(Permission.MR_MERGE, "foo") == "pgw.mr.merge.foo" + assert permission_key(Permission.HOOKS, "foo") == "pgw.hooks.foo" + + +def test_parse_mr_scoped_keys() -> None: + create = parse_permission_key("pgw.mr.create.demo") + assert create is not None + assert create.permission is Permission.MR_CREATE + merge = parse_permission_key("pgw.mr.merge.demo") + assert merge is not None + assert merge.permission is Permission.MR_MERGE + + +def test_project_glob() -> None: + assert project_matches_scope("team/app", "team/*") + assert not project_matches_scope("other/app", "team/*") + assert project_matches_scope("anything", "*") + + +def test_superuser_grants_all() -> None: + m = PermissionsMap.model_validate({ + "*": ["admin@example.com"], + "pgw.tasks.secret": ["other@example.com"], + }) + assert m.has_permission("admin@example.com", Permission.TASKS, "secret") + assert m.has_permission("admin@example.com", Permission.SETTINGS) + assert m.has_permission("other@example.com", Permission.TASKS, "secret") + assert not m.has_permission("other@example.com", Permission.SETTINGS) + assert not m.has_permission("other@example.com", Permission.TASKS, "public") + + +def test_scoped_grant() -> None: + m = PermissionsMap.model_validate({ + "*": [], + "pgw.comments.proj-a": ["reader@example.com"], + "pgw.boards.team/*": ["lead@example.com"], + }) + assert m.has_permission("reader@example.com", Permission.COMMENTS, "proj-a") + assert not m.has_permission("reader@example.com", Permission.TASKS, "proj-a") + assert m.has_permission("lead@example.com", Permission.BOARDS, "team/widget") + assert not m.has_permission("lead@example.com", Permission.BOARDS, "other/widget") + + +def test_email_case_insensitive() -> None: + m = PermissionsMap.model_validate({"*": [], PERMISSION_SETTINGS: ["Admin@Example.COM"]}) + assert m.has_permission("admin@example.com", Permission.SETTINGS) + + +def test_rejects_unknown_key() -> None: + with pytest.raises(ValueError, match="unknown permission key"): + PermissionsMap.model_validate({"pgw.notreal": ["a@b.com"]}) + + +def test_ensures_wildcard_key() -> None: + m = PermissionsMap.model_validate({PERMISSION_ADD_PROJECTS: ["u@x.com"]}) + assert "*" in m.root diff --git a/pygitweb/settings.py b/pygitweb/settings.py index 9c1b5ab..a903c44 100644 --- a/pygitweb/settings.py +++ b/pygitweb/settings.py @@ -14,9 +14,10 @@ import pygit2 from fastapi import APIRouter, Depends, Request from fastapi.responses import HTMLResponse, RedirectResponse -from pygitweb.auth import require_active_user_if_auth_enabled +from pygitweb.auth import require_permission from pygitweb.config import settings from pygitweb.dependencies import ValidatedSettingsProject +from pygitweb.permissions import Permission from pygitweb.templates_env import POSTAMBLE, PREAMBLE, env, jinja_escape @@ -32,7 +33,7 @@ GIT_CONFIG_LEVEL_WORKTREE = 6 router = APIRouter( prefix="/settings", tags=["settings"], - dependencies=[Depends(require_active_user_if_auth_enabled)], + dependencies=[Depends(require_permission(Permission.SETTINGS))], ) diff --git a/pygitweb/settings_permissions_test.py b/pygitweb/settings_permissions_test.py new file mode 100644 index 0000000..6ce7cff --- /dev/null +++ b/pygitweb/settings_permissions_test.py @@ -0,0 +1,76 @@ +from collections.abc import Generator +from unittest.mock import patch + +import pytest +from fastapi.testclient import TestClient + +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.main import app +from pygitweb.permissions import PERMISSION_SETTINGS, PermissionsMap + +_SETTINGS_AUTH = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="admin", password="secret"), LocalUser(user="viewer", password="secret")], + oauth_permissions=PermissionsMap.model_validate({ + "*": [], + PERMISSION_SETTINGS: ["admin"], + }), +) + + +@pytest.fixture +def client() -> Generator[TestClient, None, None]: + with TestClient(app) as c: + yield c + + +@pytest.fixture +def settings_auth() -> Generator[AuthConfig, None, None]: + with ( + patch("pygitweb.auth_config.auth_config", _SETTINGS_AUTH), + patch("pygitweb.auth.auth_config", _SETTINGS_AUTH), + patch("pygitweb.main.auth_config", _SETTINGS_AUTH), + ): + yield _SETTINGS_AUTH + + +def _login(client: TestClient, username: str) -> None: + client.post( + "/login", + data={"username": username, "password": "secret", "next": "/"}, + follow_redirects=False, + ) + + +def test_settings_open_when_auth_disabled(client: TestClient) -> None: + from pygitweb.config import settings + + with patch.object(settings, "AUTH", False): + r = client.get("/settings/pygitweb") + assert r.status_code == 200 + + +def test_settings_forbidden_without_login(client: TestClient, settings_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with patch.object(settings, "AUTH", True): + r = client.get("/settings/pygitweb") + assert r.status_code == 401 + + +def test_settings_allowed_with_grant(client: TestClient, settings_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with patch.object(settings, "AUTH", True): + _login(client, "admin") + r = client.get("/settings/pygitweb") + assert r.status_code == 200 + + +def test_settings_forbidden_without_grant(client: TestClient, settings_auth: AuthConfig) -> None: + from pygitweb.config import settings + + with patch.object(settings, "AUTH", True): + _login(client, "viewer") + r = client.get("/settings/pygit2") + assert r.status_code == 403 diff --git a/pygitweb/tasks.py b/pygitweb/tasks.py index dd5d946..51754b0 100644 --- a/pygitweb/tasks.py +++ b/pygitweb/tasks.py @@ -24,9 +24,10 @@ from pygittools.tasks import ( get_task, get_task_by_oid, ) -from pygitweb.auth import require_active_user_if_auth_enabled +from pygitweb.auth import require_active_user_if_auth_enabled, require_permission from pygitweb.config import settings from pygitweb.dependencies import ValidatedQueryProject +from pygitweb.permissions import Permission # Well-known empty tree OID for boards/tasks when repo has no HEAD EMPTY_TREE_OID = pygit2.Oid(hex="4b825dc642cb6eb9a060e54bf8d69288fbee4904") @@ -41,9 +42,10 @@ def _task_ref(task_id: str) -> str: def _repo_head_or_empty(repo: pygit2.Repository) -> pygit2.Oid | str: - if repo.head: + try: return repo.head.target - return EMPTY_TREE_OID + except KeyError: + return EMPTY_TREE_OID def create_board_for_project( @@ -64,10 +66,14 @@ def create_board_for_project( # ---------- Board Routes ---------- -board_router = APIRouter(tags=["boards"], dependencies=[Depends(require_active_user_if_auth_enabled)]) +board_router = APIRouter(tags=["boards"]) -@board_router.get("/list", response_class=JSONResponse) +@board_router.get( + "/list", + response_class=JSONResponse, + dependencies=[Depends(require_active_user_if_auth_enabled)], +) def boards_list(project: ValidatedQueryProject) -> JSONResponse: """List all boards for the project.""" repo = pygit2.Repository(os.path.join(settings.PROJECTROOT, project)) @@ -91,7 +97,11 @@ def boards_list(project: ValidatedQueryProject) -> JSONResponse: return JSONResponse(content=boards) -@board_router.post("/create", response_class=JSONResponse) +@board_router.post( + "/create", + response_class=JSONResponse, + dependencies=[Depends(require_permission(Permission.BOARDS, project_from_query=True))], +) def boards_create( project: ValidatedQueryProject, name: str = Query(..., description="Board name"), @@ -110,7 +120,11 @@ def boards_create( return JSONResponse(content={"name": name, "ref": ref, "oid": str(oid)}) -@board_router.post("/delete", response_class=JSONResponse) +@board_router.post( + "/delete", + response_class=JSONResponse, + dependencies=[Depends(require_permission(Permission.BOARDS, project_from_query=True))], +) def boards_delete( project: ValidatedQueryProject, name: str = Query(..., description="Board name"), @@ -126,7 +140,7 @@ def boards_delete( # ---------- Task Routes ---------- -task_router = APIRouter(tags=["tasks"], dependencies=[Depends(require_active_user_if_auth_enabled)]) +task_router = APIRouter(tags=["tasks"]) def _task_to_json(t: Task) -> dict[str, Any]: @@ -184,7 +198,11 @@ def get_board_tasks_grouped( ] -@task_router.get("/list", response_class=JSONResponse) +@task_router.get( + "/list", + response_class=JSONResponse, + dependencies=[Depends(require_active_user_if_auth_enabled)], +) def task_list( project: ValidatedQueryProject, board: str = Query(..., description="Board name"), @@ -208,7 +226,11 @@ def task_list( return JSONResponse(content=tasks) -@task_router.post("/create", response_class=JSONResponse) +@task_router.post( + "/create", + response_class=JSONResponse, + dependencies=[Depends(require_permission(Permission.TASKS, project_from_query=True))], +) def task_create( project: ValidatedQueryProject, board: str = Query(..., description="Board name"), @@ -257,7 +279,11 @@ def task_create( return JSONResponse(content={"task_id": task_id, "ref": ref, "oid": str(oid)}) -@task_router.post("/update", response_class=JSONResponse) +@task_router.post( + "/update", + response_class=JSONResponse, + dependencies=[Depends(require_permission(Permission.TASKS, project_from_query=True))], +) async def task_update( request: Request, project: ValidatedQueryProject, @@ -310,7 +336,11 @@ async def task_update( return JSONResponse(content=_task_to_json(t)) -@task_router.post("/delete", response_class=JSONResponse) +@task_router.post( + "/delete", + response_class=JSONResponse, + dependencies=[Depends(require_permission(Permission.TASKS, project_from_query=True))], +) def task_delete( project: ValidatedQueryProject, board: str = Query(..., description="Board name"), @@ -337,7 +367,7 @@ def task_delete( # ---------- Comment Routes ---------- -comment_router = APIRouter(tags=["comments"], dependencies=[Depends(require_active_user_if_auth_enabled)]) +comment_router = APIRouter(tags=["comments"]) def _comment_to_json(c: Comment) -> dict[str, Any]: @@ -349,7 +379,11 @@ def _comment_to_json(c: Comment) -> dict[str, Any]: } -@comment_router.get("/list", response_class=JSONResponse) +@comment_router.get( + "/list", + response_class=JSONResponse, + dependencies=[Depends(require_active_user_if_auth_enabled)], +) def comment_list( project: ValidatedQueryProject, board: str = Query(..., description="Board name"), @@ -374,7 +408,11 @@ def comment_list( return JSONResponse(content=comments) -@comment_router.post("/create", response_class=JSONResponse) +@comment_router.post( + "/create", + response_class=JSONResponse, + dependencies=[Depends(require_permission(Permission.COMMENTS, project_from_query=True))], +) def comment_create( project: ValidatedQueryProject, board: str = Query(..., description="Board name"), @@ -387,8 +425,10 @@ def comment_create( t = get_task(repo, task_ref) if not t: raise HTTPException(status_code=404, detail="Task not found") - resolved = repo.revparse_single(task_ref) - target_oid = resolved.oid if hasattr(resolved, "oid") else pygit2.Oid(hex=str(resolved)) + try: + target_oid = repo.references[task_ref].resolve().target + except KeyError as exc: + raise HTTPException(status_code=404, detail="Task not found") from exc comment = Comment(target=target_oid, tagger="", content=content or "") comment_oid = comment.write(repo) t.comments = getattr(t, "comments", []) or [] @@ -398,7 +438,11 @@ def comment_create( return JSONResponse(content={"oid": str(comment_oid), "message": "Comment created"}) -@comment_router.post("/update", response_class=JSONResponse) +@comment_router.post( + "/update", + response_class=JSONResponse, + dependencies=[Depends(require_active_user_if_auth_enabled)], +) async def comment_update( request: Request, project: ValidatedQueryProject, @@ -438,7 +482,11 @@ async def comment_update( ) -@comment_router.post("/delete", response_class=JSONResponse) +@comment_router.post( + "/delete", + response_class=JSONResponse, + dependencies=[Depends(require_active_user_if_auth_enabled)], +) def comment_delete( project: ValidatedQueryProject, board: str = Query(..., description="Board name"), diff --git a/pygitweb/tasks_permissions_test.py b/pygitweb/tasks_permissions_test.py new file mode 100644 index 0000000..cf09eb0 --- /dev/null +++ b/pygitweb/tasks_permissions_test.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +from collections.abc import Generator +from pathlib import Path +from unittest.mock import patch + +import pygit2 +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from pygittools.tasks import BOARD_REF_PREFIX, Board +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.config import settings +from pygitweb.conftest import client_as_user +from pygitweb.permissions import Permission, PermissionsMap, permission_key +from pygitweb.tasks import EMPTY_TREE_OID, task_router + +_TASKS_AUTH = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="admin", password="secret"), LocalUser(user="viewer", password="secret")], + oauth_permissions=PermissionsMap.model_validate({ + "*": [], + permission_key(Permission.TASKS, "demo"): ["admin"], + }), +) + + +def _build_client() -> TestClient: + app = FastAPI() + app.include_router(task_router, prefix="/tasks") + return TestClient(app) + + +def _create_initial_commit(repo: pygit2.Repository, repo_dir: Path) -> None: + (repo_dir / "README.md").write_text("seed\n", encoding="utf-8") + index = repo.index + index.add("README.md") + index.write() + tree = index.write_tree() + sig = pygit2.Signature("tester", "tester@example.com") + repo.create_commit("HEAD", sig, sig, "initial", tree, []) + + +@pytest.fixture +def task_env(tmp_path: Path) -> Generator[dict[str, str | TestClient], None, None]: + repo_dir = tmp_path / "demo" + repo = pygit2.init_repository(str(repo_dir), bare=False) + _create_initial_commit(repo, repo_dir) + board = Board(EMPTY_TREE_OID, f"{BOARD_REF_PREFIX}Tasks", tagger="", description="") + board.write(repo) + with ( + patch.object(settings, "PROJECTROOT", str(tmp_path)), + patch.object(settings, "PROJECTS_LIST", str(tmp_path)), + patch.object(settings, "STRICT_EXPORT", False), + patch.object(settings, "EXPORT_OK", ""), + patch.object(settings, "LIST_ALL", True), + patch("pygitweb.auth_config.auth_config", _TASKS_AUTH), + patch("pygitweb.auth.auth_config", _TASKS_AUTH), + ): + yield {"client": _build_client(), "project": "demo", "board": "Tasks"} + + +def test_task_post_open_when_auth_disabled(task_env: dict[str, str | TestClient]) -> None: + client = task_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", False): + r = client.post( + "/tasks/create", + params={ + "project": task_env["project"], + "board": task_env["board"], + "title": "New task", + }, + ) + assert r.status_code == 200 + + +def test_task_create_401_without_login(task_env: dict[str, str | TestClient]) -> None: + client = task_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True): + r = client.post( + "/tasks/create", + params={ + "project": task_env["project"], + "board": task_env["board"], + "title": "New task", + }, + ) + assert r.status_code == 401 + + +def test_task_create_allowed_with_grant(task_env: dict[str, str | TestClient]) -> None: + client = task_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "admin"): + r = client.post( + "/tasks/create", + params={ + "project": task_env["project"], + "board": task_env["board"], + "title": "Granted task", + }, + ) + assert r.status_code == 200 + + +def test_task_create_forbidden_without_grant(task_env: dict[str, str | TestClient]) -> None: + client = task_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "viewer"): + r = client.post( + "/tasks/create", + params={ + "project": task_env["project"], + "board": task_env["board"], + "title": "Denied task", + }, + ) + assert r.status_code == 403 + + +def test_task_update_forbidden_without_grant(task_env: dict[str, str | TestClient]) -> None: + client = task_env["client"] + assert isinstance(client, TestClient) + with patch.object(settings, "AUTH", True), client_as_user(client, "viewer"): + r = client.post( + "/tasks/update", + params={ + "project": task_env["project"], + "board": task_env["board"], + "task": "refs/tags/tasks/task_missing", + }, + json={"title": "nope"}, + ) + assert r.status_code == 403