diff --git a/pygitweb/auth.schema.json b/pygitweb/auth.schema.json index cf91728..9792b19 100644 --- a/pygitweb/auth.schema.json +++ b/pygitweb/auth.schema.json @@ -60,7 +60,7 @@ "additionalProperties": { "type": "array", "items": { "type": "string" }, - "description": "Scoped keys: pgw.read.{scope}, pgw.write.{scope}, pgw.write_branch.{project/branch} (refs/heads only), pgw.write_ref.{project/path-after-refs}, pgw.boards.{scope}, pgw.tasks.{scope}, pgw.comments.{scope}, pgw.hooks.{scope}, pgw.mr.create.{scope}, pgw.mr.merge.{scope} (fnmatch). Principal * matches any authenticated user." + "description": "Scoped keys: pgw.read.{scope}, pgw.write.{scope}, pgw.write_ref.{project/path-after-refs} (whitelist when any write_ref applies to the project; otherwise project write covers all refs), pgw.boards.{scope}, pgw.tasks.{scope}, pgw.comments.{scope}, pgw.hooks.{scope}, pgw.mr.create.{scope}, pgw.mr.merge.{scope} (fnmatch). Principal * matches any authenticated user." } }, "local_users": { diff --git a/pygitweb/src/pygitweb/auth.py b/pygitweb/src/pygitweb/auth.py index 940f0da..3c3db35 100644 --- a/pygitweb/src/pygitweb/auth.py +++ b/pygitweb/src/pygitweb/auth.py @@ -45,8 +45,7 @@ from pygitweb.password_hash import hash_password, verify_password from pygitweb.permissions import ( Permission, PermissionPrincipal, - ref_permission_subject, - ref_write_grant, + has_ref_write_permission, ) from pygitweb.projects import git_get_project_owner from pygitweb.sessions import AuthMethod, create_session, get_session, revoke_session @@ -190,22 +189,23 @@ def can_write_project(project: str, token: str | None) -> bool: def can_push_ref(project: str, principal: PermissionPrincipal, ref_name: str) -> bool: - """Per-ref push check. Project write is the fallback. + """Per-ref push check. - ``pgw.write_branch`` matches ``refs/heads`` only. ``pgw.write_ref`` matches any - ref under ``refs/``. A name outside ``refs/`` is rejected. When no specific - grant matches, project write (including ownership) is enough. + Project owners may push any ref. Otherwise, if any ``pgw.write_ref`` rule + applies to the project, only matching ``write_ref`` grants allow the push + (whitelist). With no ``write_ref`` rules for the project, project write is + enough. Names outside ``refs/`` are rejected. """ if not settings.AUTH: return True - if ref_permission_subject(project, ref_name) is None: - return False - permissions = auth_config.oauth_permissions - grant = ref_write_grant(permissions, project, ref_name) - if grant is not None: - permission, subject = grant - return permissions.has_permission(principal.identity, permission, subject) - return principal_can_write_project(project, principal) + if _owner_matches_principal(project, principal): + return True + return has_ref_write_permission( + auth_config.oauth_permissions, + principal.identity, + project, + ref_name, + ) def ensure_read_project(project: str, token: str | None) -> None: diff --git a/pygitweb/src/pygitweb/permissions.py b/pygitweb/src/pygitweb/permissions.py index 1694984..8cc19e5 100644 --- a/pygitweb/src/pygitweb/permissions.py +++ b/pygitweb/src/pygitweb/permissions.py @@ -7,10 +7,14 @@ every permission. Scoped keys use a project segment that supports ``fnmatch`` globs (``*``, ``?``, ``[seq]``), e.g. ``pgw.tasks.foo/*``, ``pgw.read.*``, or ``pgw.write.myproj``. -``pgw.write_branch`` scopes are ``project/branch`` and apply only to ``refs/heads``. ``pgw.write_ref`` scopes are ``project`` plus the path after ``refs/`` (for example ``demo/heads/main``, ``demo/tags/v1.*``, ``demo/notes/*``). +When any ``pgw.write_ref`` key applies to a project, those grants are a whitelist: +only listed principals may push matching refs, and project write alone is not +enough for any ref. When no ``write_ref`` key applies to the project, project +write remains sufficient for every ref under ``refs/``. + Principal ``*`` in a grant list matches any authenticated principal. """ @@ -31,11 +35,10 @@ PERMISSION_ADD_PROJECTS = "pgw.addprojects" PERMISSION_SETTINGS = "pgw.settings" _SCOPED_PREFIX = re.compile( - r"^pgw\.(boards|tasks|comments|hooks|mr\.create|mr\.merge|read|write_branch|write_ref|write)\.(.+)$", + r"^pgw\.(boards|tasks|comments|hooks|mr\.create|mr\.merge|read|write_ref|write)\.(.+)$", ) _REFS_PREFIX = "refs/" -_BRANCH_REF_PREFIX = "refs/heads/" class Permission(StrEnum): @@ -43,7 +46,6 @@ class Permission(StrEnum): READ = "read" WRITE = "write" - WRITE_BRANCH = "write_branch" WRITE_REF = "write_ref" BOARDS = "boards" TASKS = "tasks" @@ -67,7 +69,6 @@ class Permission(StrEnum): if self in ( Permission.READ, Permission.WRITE, - Permission.WRITE_BRANCH, Permission.WRITE_REF, Permission.BOARDS, Permission.TASKS, @@ -128,17 +129,10 @@ def project_matches_scope(project: str, scope: str) -> bool: return fnmatch.fnmatchcase(project, scope) -def short_branch_name_from_ref(ref_name: str) -> str | None: - if not ref_name.startswith(_BRANCH_REF_PREFIX): - return None - return ref_name.removeprefix(_BRANCH_REF_PREFIX) - - -def branch_permission_subject(project: str, ref_name: str) -> str | None: - short_name = short_branch_name_from_ref(ref_name) - if short_name is None: - return None - return f"{project}/{short_name}" +def write_ref_project_segment(scope: str) -> str: + """Project segment of a ``write_ref`` scope (text before the first ``/``).""" + project_part, sep, _rest = scope.partition("/") + return project_part if sep else scope def ref_permission_subject(project: str, ref_name: str) -> str | None: @@ -150,22 +144,17 @@ def ref_permission_subject(project: str, ref_name: str) -> str | None: return f"{project}/{rest}" -def ref_write_grant( - permissions: PermissionsMap, - project: str, - ref_name: str, -) -> tuple[Permission, str] | None: - """Specific ``write_branch`` or ``write_ref`` rule for this ref, if one matches. - - ``pgw.write_branch`` is checked first and only for ``refs/heads/*``. - """ - legacy = branch_permission_subject(project, ref_name) - if legacy is not None and permissions.matching_scoped_principals(Permission.WRITE_BRANCH, legacy): - return Permission.WRITE_BRANCH, legacy - subject = ref_permission_subject(project, ref_name) - if subject is not None and permissions.matching_scoped_principals(Permission.WRITE_REF, subject): - return Permission.WRITE_REF, subject - return None +def project_has_write_ref_rules(permissions: PermissionsMap, project: str) -> bool: + """True when any ``pgw.write_ref`` key's project segment matches ``project``.""" + for key in permissions.root: + if key == WILDCARD_PERMISSION_KEY: + continue + parsed = parse_permission_key(key) + if parsed is None or parsed.permission is not Permission.WRITE_REF or parsed.project_scope is None: + continue + if project_matches_scope(project, write_ref_project_segment(parsed.project_scope)): + return True + return False def has_ref_write_permission( @@ -174,12 +163,16 @@ def has_ref_write_permission( project: str, ref_name: str, ) -> bool: - if ref_permission_subject(project, ref_name) is None: + """Whether ``principal`` may push ``ref_name`` in ``project``. + + If any ``write_ref`` rule applies to ``project``, only matching ``write_ref`` + grants allow the push (whitelist). Otherwise project ``write`` is enough. + """ + subject = ref_permission_subject(project, ref_name) + if subject is None: return False - grant = ref_write_grant(permissions, project, ref_name) - if grant is not None: - permission, subject = grant - return permissions.has_permission(principal, permission, subject) + if project_has_write_ref_rules(permissions, project): + return permissions.has_permission(principal, Permission.WRITE_REF, subject) return permissions.has_permission(principal, Permission.WRITE, project) diff --git a/pygitweb/src/pygitweb/permissions_test.py b/pygitweb/src/pygitweb/permissions_test.py index 54c95d8..ebe061a 100644 --- a/pygitweb/src/pygitweb/permissions_test.py +++ b/pygitweb/src/pygitweb/permissions_test.py @@ -5,13 +5,13 @@ from pygitweb.permissions import ( PERMISSION_SETTINGS, Permission, PermissionsMap, - branch_permission_subject, has_ref_write_permission, parse_permission_key, permission_key, + project_has_write_ref_rules, project_matches_scope, ref_permission_subject, - short_branch_name_from_ref, + write_ref_project_segment, ) @@ -58,15 +58,16 @@ def test_parse_mr_scoped_keys() -> None: assert merge.permission is Permission.MR_MERGE -def test_parse_branch_scoped_keys() -> None: - write = parse_permission_key("pgw.write_branch.dev/alice.*") - assert write is not None - assert write.permission is Permission.WRITE_BRANCH - assert write.project_scope == "dev/alice.*" +def test_parse_write_ref_scoped_keys() -> None: ref = parse_permission_key("pgw.write_ref.demo/tags/v*") assert ref is not None assert ref.permission is Permission.WRITE_REF assert ref.project_scope == "demo/tags/v*" + heads = parse_permission_key("pgw.write_ref.dev/heads/alice.*") + assert heads is not None + assert heads.permission is Permission.WRITE_REF + assert heads.project_scope == "dev/heads/alice.*" + assert parse_permission_key("pgw.write_branch.dev/alice.*") is None assert parse_permission_key("pgw.read_branch.team/app-main") is None @@ -76,16 +77,14 @@ def test_project_glob() -> None: assert project_matches_scope("anything", "*") -def test_branch_permission_subject() -> None: - assert short_branch_name_from_ref("refs/heads/main") == "main" - assert short_branch_name_from_ref("refs/heads/dev/alice.feature") == "dev/alice.feature" - assert short_branch_name_from_ref("refs/tags/v1.0.0") is None - assert branch_permission_subject("demo", "refs/heads/main") == "demo/main" - assert branch_permission_subject("demo", "refs/tags/v1.0.0") is None +def test_ref_permission_subject() -> None: assert ref_permission_subject("demo", "refs/heads/main") == "demo/heads/main" assert ref_permission_subject("demo", "refs/tags/v1.0.0") == "demo/tags/v1.0.0" assert ref_permission_subject("demo", "refs/notes/commits") == "demo/notes/commits" assert ref_permission_subject("demo", "HEAD") is None + assert write_ref_project_segment("demo/heads/main") == "demo" + assert write_ref_project_segment("*/tags/v*") == "*" + assert write_ref_project_segment("solo") == "solo" def test_superuser_grants_all() -> None: @@ -120,6 +119,8 @@ def test_email_case_insensitive() -> None: def test_rejects_unknown_key() -> None: with pytest.raises(ValueError, match="unknown permission key"): PermissionsMap.model_validate({"pgw.notreal": ["a@b.com"]}) + with pytest.raises(ValueError, match="unknown permission key"): + PermissionsMap.model_validate({"pgw.write_branch.demo/main": ["a@b.com"]}) def test_ensures_wildcard_key() -> None: @@ -173,6 +174,7 @@ def test_has_ref_write_permission_falls_back_to_project_scope() -> None: "*": [], "pgw.write.demo": ["writer@example.com"], }) + assert not project_has_write_ref_rules(m, "demo") assert has_ref_write_permission(m, "writer@example.com", "demo", "refs/heads/main") assert has_ref_write_permission(m, "writer@example.com", "demo", "refs/tags/v1") assert has_ref_write_permission(m, "writer@example.com", "demo", "refs/notes/commits") @@ -180,39 +182,57 @@ def test_has_ref_write_permission_falls_back_to_project_scope() -> None: assert not has_ref_write_permission(m, "writer@example.com", "demo", "HEAD") -def test_has_ref_write_permission_enforces_branch_specific_rules() -> None: +def test_write_ref_rules_are_per_project() -> None: m = PermissionsMap.model_validate({ "*": [], "pgw.write.demo": ["writer@example.com"], - "pgw.write_branch.demo/main": ["branch-writer@example.com"], + "pgw.write.other": ["writer@example.com"], + "pgw.write_ref.other/heads/main": ["ref-writer@example.com"], }) - assert has_ref_write_permission(m, "branch-writer@example.com", "demo", "refs/heads/main") - assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/heads/main") - assert has_ref_write_permission(m, "writer@example.com", "demo", "refs/tags/v1") + assert not project_has_write_ref_rules(m, "demo") + assert project_has_write_ref_rules(m, "other") + assert has_ref_write_permission(m, "writer@example.com", "demo", "refs/heads/main") + assert not has_ref_write_permission(m, "writer@example.com", "other", "refs/heads/main") + assert has_ref_write_permission(m, "ref-writer@example.com", "other", "refs/heads/main") -def test_has_ref_write_permission_enforces_ref_rules() -> None: +def test_has_ref_write_permission_whitelist_when_write_ref_defined() -> None: m = PermissionsMap.model_validate({ "*": [], "pgw.write.demo": ["writer@example.com"], "pgw.write_ref.demo/tags/v*": ["release@example.com"], "pgw.write_ref.demo/notes/*": ["alice@example.com"], "pgw.write_ref.demo/heads/main": ["ref-writer@example.com"], + "pgw.write_ref.demo/heads/feature*": ["writer@example.com"], }) + assert project_has_write_ref_rules(m, "demo") assert has_ref_write_permission(m, "release@example.com", "demo", "refs/tags/v1.2") assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/tags/v1.2") - assert has_ref_write_permission(m, "writer@example.com", "demo", "refs/tags/nightly") + assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/tags/nightly") assert has_ref_write_permission(m, "alice@example.com", "demo", "refs/notes/commits") assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/notes/commits") assert has_ref_write_permission(m, "ref-writer@example.com", "demo", "refs/heads/main") assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/heads/main") + assert has_ref_write_permission(m, "writer@example.com", "demo", "refs/heads/feature/x") + assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/heads/develop") -def test_write_branch_overrides_write_ref_for_heads() -> None: +def test_write_ref_whitelist_empty_grant_list_still_enables_mode() -> None: m = PermissionsMap.model_validate({ "*": [], - "pgw.write_branch.demo/main": ["branch-writer@example.com"], + "pgw.write.demo": ["writer@example.com"], + "pgw.write_ref.demo/heads/main": [], + }) + assert project_has_write_ref_rules(m, "demo") + assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/heads/main") + assert not has_ref_write_permission(m, "writer@example.com", "demo", "refs/tags/v1") + + +def test_write_ref_superuser_bypasses_whitelist() -> None: + m = PermissionsMap.model_validate({ + "*": ["admin@example.com"], + "pgw.write.demo": ["writer@example.com"], "pgw.write_ref.demo/heads/main": ["ref-writer@example.com"], }) - assert has_ref_write_permission(m, "branch-writer@example.com", "demo", "refs/heads/main") - assert not has_ref_write_permission(m, "ref-writer@example.com", "demo", "refs/heads/main") + assert has_ref_write_permission(m, "admin@example.com", "demo", "refs/heads/main") + assert has_ref_write_permission(m, "admin@example.com", "demo", "refs/tags/v1") diff --git a/pygitweb/src/pygitweb/smart_http_test.py b/pygitweb/src/pygitweb/smart_http_test.py index 4d1c341..cce9c5b 100644 --- a/pygitweb/src/pygitweb/smart_http_test.py +++ b/pygitweb/src/pygitweb/smart_http_test.py @@ -428,18 +428,19 @@ class TestSmartHttpRoutes: assert fetch.status_code == 404 assert push.status_code == 403 - def test_branch_push_permission_on_receive_pack(self, http_repo_env: dict[str, str]) -> None: + def test_write_ref_whitelist_on_receive_pack(self, http_repo_env: dict[str, str]) -> None: restricted = PermissionsMap.model_validate({ "*": [], "pgw.read.*": ["*"], "pgw.write.*": ["*"], - "pgw.write_branch.demo/main": ["other@example.com"], + "pgw.write_ref.demo/heads/main": ["other@example.com"], }) allowed = PermissionsMap.model_validate({ "*": [], "pgw.read.*": ["*"], "pgw.write.*": ["*"], - "pgw.write_branch.demo/main": ["alice@example.com"], + "pgw.write_ref.demo/heads/main": ["alice@example.com"], + "pgw.write_ref.demo/tags/*": ["alice@example.com"], }) client = TestClient(app) denied_body = _push_commands("refs/heads/main") @@ -459,8 +460,9 @@ class TestSmartHttpRoutes: ) assert denied.status_code == 403 assert "refs/heads/main" in denied.json()["detail"] - assert tag.status_code == 200 - assert run.call_count == 1 + assert tag.status_code == 403 + assert "refs/tags/v1" in tag.json()["detail"] + assert run.call_count == 0 with ( _auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": allowed})), patch("pygitweb.smart_http._run_git_service", return_value=b"") as run_allowed, @@ -470,8 +472,14 @@ class TestSmartHttpRoutes: content=denied_body, auth=("alice", _WRITE_KEY), ) + tag_ok = client.post( + f"/http/{http_repo_env['project']}/git-receive-pack", + content=_push_commands("refs/tags/v1"), + auth=("alice", _WRITE_KEY), + ) assert accepted.status_code == 200 - assert run_allowed.call_count == 1 + assert tag_ok.status_code == 200 + assert run_allowed.call_count == 2 with ( _auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": allowed})), patch("pygitweb.smart_http._run_git_service", return_value=b""), @@ -486,6 +494,31 @@ class TestSmartHttpRoutes: ) assert oauth_push.status_code == 200 + def test_no_write_ref_falls_back_to_project_write(self, http_repo_env: dict[str, str]) -> None: + open_write = PermissionsMap.model_validate({ + "*": [], + "pgw.read.*": ["*"], + "pgw.write.*": ["*"], + }) + client = TestClient(app) + with ( + _auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": open_write})), + patch("pygitweb.smart_http._run_git_service", return_value=b"") as run, + ): + main = client.post( + f"/http/{http_repo_env['project']}/git-receive-pack", + content=_push_commands("refs/heads/main"), + auth=("alice", _WRITE_KEY), + ) + tag = client.post( + f"/http/{http_repo_env['project']}/git-receive-pack", + content=_push_commands("refs/tags/v1"), + auth=("alice", _WRITE_KEY), + ) + assert main.status_code == 200 + assert tag.status_code == 200 + assert run.call_count == 2 + def test_ref_push_permission_on_receive_pack(self, http_repo_env: dict[str, str]) -> None: restricted = PermissionsMap.model_validate({ "*": [], @@ -514,11 +547,17 @@ class TestSmartHttpRoutes: content=_push_commands("refs/tags/nightly"), auth=("alice", _WRITE_KEY), ) + branch = client.post( + f"/http/{http_repo_env['project']}/git-receive-pack", + content=_push_commands("refs/heads/main"), + auth=("alice", _WRITE_KEY), + ) assert tag.status_code == 403 assert "refs/tags/v1" in tag.json()["detail"] assert notes.status_code == 403 - assert other_tag.status_code == 200 - assert run.call_count == 1 + assert other_tag.status_code == 403 + assert branch.status_code == 403 + assert run.call_count == 0 def test_clone_and_push_over_http(self, http_repo_env: dict[str, str]) -> None: with _http_server() as port: