diff --git a/config/repo_config.py b/config/repo_config.py index 4bc107455d..66147ce932 100644 --- a/config/repo_config.py +++ b/config/repo_config.py @@ -32,6 +32,7 @@ """ import os +import time from pathlib import Path from typing import Any, cast @@ -343,6 +344,73 @@ def get_checkpoint_repo(repo: str) -> str | None: return cast(str | None, get_repo_setting(repo, "checkpoint_repo", None)) +_checkpoint_repos_cache: tuple[float, frozenset[str]] | None = None +_CHECKPOINT_REPOS_TTL = 60 # seconds + + +def get_all_checkpoint_repos() -> frozenset[str]: + """Get the set of all configured checkpoint repositories. + + Scans all repo_settings entries and collects every checkpoint_repo value. + Used by the gateway to exempt checkpoint repos from private mode policy. + + Results are cached for 60 seconds to avoid redundant config file I/O + on every git request. + + Returns: + Frozenset of checkpoint repo names in "owner/repo" format, lowercased. + Returns empty frozenset if config cannot be loaded or has no checkpoint repos. + """ + global _checkpoint_repos_cache + + now = time.monotonic() + if _checkpoint_repos_cache is not None: + cached_time, cached_result = _checkpoint_repos_cache + if now - cached_time < _CHECKPOINT_REPOS_TTL: + return cached_result + + try: + config = _load_config() + except Exception: + result: frozenset[str] = frozenset() + _checkpoint_repos_cache = (now, result) + return result + + repo_settings = config.get("repo_settings", {}) + if not isinstance(repo_settings, dict): + result = frozenset() + _checkpoint_repos_cache = (now, result) + return result + + repos: set[str] = set() + for settings in repo_settings.values(): + if isinstance(settings, dict): + checkpoint_repo = settings.get("checkpoint_repo") + if checkpoint_repo and isinstance(checkpoint_repo, str): + repos.add(checkpoint_repo.lower()) + result = frozenset(repos) + _checkpoint_repos_cache = (now, result) + return result + + +def is_checkpoint_repo(owner: str, repo: str) -> bool: + """Check if a repository is configured as a checkpoint destination. + + Args: + owner: Repository owner (e.g. "jwbron") + repo: Repository name (e.g. "egg-checkpoints") + + Returns: + True if owner/repo is a configured checkpoint_repo. + False on any config error (fail-closed). + """ + try: + full_name = f"{owner}/{repo}".lower() + return full_name in get_all_checkpoint_repos() + except Exception: + return False + + def get_repo_checks(repo: str) -> list[dict[str, str]]: """Get configured check commands for a repository. diff --git a/gateway/gateway.py b/gateway/gateway.py index 2382121683..ce3a714697 100644 --- a/gateway/gateway.py +++ b/gateway/gateway.py @@ -183,7 +183,7 @@ _config_path = Path(__file__).parent.parent / "config" if _config_path.exists() and str(_config_path) not in sys.path: sys.path.insert(0, str(_config_path)) -from repo_config import get_auth_mode, get_checkpoint_repo +from repo_config import get_auth_mode, get_checkpoint_repo, is_checkpoint_repo logger = get_logger("gateway") @@ -569,31 +569,47 @@ def git_push() -> tuple[Response, int] | Response: repo_info = parse_owner_repo(repo) if repo_info: - priv_result = check_private_repo_access( - operation="push", - owner=repo_info.owner, - repo=repo_info.repo, - for_write=True, - session_mode=session_mode, - ) - if not priv_result.allowed: + # Checkpoint operations are infrastructure — always accessible regardless of + # session mode. This covers both dedicated checkpoint repos and checkpoint + # branch pushes to the source repo itself. + if is_checkpoint_push or is_checkpoint_repo(repo_info.owner, repo_info.repo): audit_log( - "push_denied_private_mode", + "push_checkpoint_exempt", "git_push", - success=False, + success=True, details={ "repo": repo, "branch": branch, - "reason": priv_result.reason, - "visibility": priv_result.visibility, - "auth_mode": auth_mode, + "reason": "Checkpoint operation exempt from private mode policy", + "exempt_type": "checkpoint_repo" if is_checkpoint_repo(repo_info.owner, repo_info.repo) else "checkpoint_branch", }, ) - return make_error( - priv_result.reason, - status_code=403, - details=priv_result.to_dict(), + else: + priv_result = check_private_repo_access( + operation="push", + owner=repo_info.owner, + repo=repo_info.repo, + for_write=True, + session_mode=session_mode, ) + if not priv_result.allowed: + audit_log( + "push_denied_private_mode", + "git_push", + success=False, + details={ + "repo": repo, + "branch": branch, + "reason": priv_result.reason, + "visibility": priv_result.visibility, + "auth_mode": auth_mode, + }, + ) + return make_error( + priv_result.reason, + status_code=403, + details=priv_result.to_dict(), + ) # Check branch ownership policy (pass auth mode for relaxed policy in user mode) policy = get_policy_engine() @@ -1408,29 +1424,41 @@ def git_fetch() -> tuple[Response, int] | Response: # Check Private Repo Mode policy (if enabled) repo_info = parse_owner_repo(repo) if repo_info: - priv_result = check_private_repo_access( - operation=operation, - owner=repo_info.owner, - repo=repo_info.repo, - for_write=False, - session_mode=session_mode, - ) - if not priv_result.allowed: + # Checkpoint repos are infrastructure — always accessible regardless of session mode + if is_checkpoint_repo(repo_info.owner, repo_info.repo): audit_log( - f"{operation}_denied_private_mode", + f"{operation}_checkpoint_repo_exempt", f"git_{operation}", - success=False, + success=True, details={ "repo": repo, - "reason": priv_result.reason, - "visibility": priv_result.visibility, + "reason": "Checkpoint repo exempt from private mode policy", }, ) - return make_error( - priv_result.reason, - status_code=403, - details=priv_result.to_dict(), + else: + priv_result = check_private_repo_access( + operation=operation, + owner=repo_info.owner, + repo=repo_info.repo, + for_write=False, + session_mode=session_mode, ) + if not priv_result.allowed: + audit_log( + f"{operation}_denied_private_mode", + f"git_{operation}", + success=False, + details={ + "repo": repo, + "reason": priv_result.reason, + "visibility": priv_result.visibility, + }, + ) + return make_error( + priv_result.reason, + status_code=403, + details=priv_result.to_dict(), + ) # Get authentication token using shared helper token_str, auth_mode, token_error = get_token_for_repo(repo) diff --git a/gateway/tests/test_gateway.py b/gateway/tests/test_gateway.py index 3c47ee6fc0..fd9b5dff17 100644 --- a/gateway/tests/test_gateway.py +++ b/gateway/tests/test_gateway.py @@ -4348,3 +4348,290 @@ def test_status_allowed_in_worktree(self, client, local_mode_headers): cmd = mock_run.call_args[0][0] assert "status" in cmd assert "--porcelain" in cmd + + +class TestCheckpointRepoBypass: + """Tests for checkpoint repo exemption from private mode policy. + + Checkpoint repos are infrastructure and should be accessible regardless + of session mode. When is_checkpoint_repo() returns True, the gateway + should skip the check_private_repo_access() call. + """ + + @pytest.fixture + def session_auth_headers(self): + """Session auth headers that do NOT patch check_private_repo_access. + + Unlike the standard auth_headers fixture, this one leaves + check_private_repo_access unpatched so we can verify the + checkpoint repo bypass logic. + """ + import sys + + import auth + + mock_session = MagicMock() + mock_session.mode = "public" + mock_session.container_id = "test-container" + mock_session.expires_at = None + mock_session.pipeline_id = None + + mock_result = SessionValidationResult(valid=True, session=mock_session) + + auth._session_manager = None + auth._rate_limiter = None + if "gateway.auth" in sys.modules: + sys.modules["gateway.auth"]._session_manager = None + sys.modules["gateway.auth"]._rate_limiter = None + + current_session_manager = sys.modules.get("session_manager", session_manager) + + with patch.object( + current_session_manager, "validate_session_for_request", return_value=mock_result + ): + yield {"Authorization": "Bearer test-session-token"} + + def _mock_subprocess_for_remote(self, operation="fetch"): + """Return a subprocess side_effect that resolves remote URLs.""" + + def run_side_effect(*args, **kwargs): + cmd = args[0] if args else kwargs.get("args", []) + result = MagicMock() + result.returncode = 0 + result.stderr = "" + + if "remote" in cmd and "get-url" in cmd: + result.stdout = "https://github.com/ckpt-owner/ckpt-repo.git\n" + elif operation in cmd: + result.stdout = "" + else: + result.stdout = "" + return result + + return run_side_effect + + def test_fetch_checkpoint_repo_bypasses_private_mode(self, client, session_auth_headers): + """Fetch on a checkpoint repo skips check_private_repo_access entirely.""" + with ( + patch("subprocess.run") as mock_run, + patch.object(gateway, "get_token_for_repo", return_value=("token", "bot", "")), + patch.object(gateway, "is_checkpoint_repo", return_value=True), + patch.object(gateway, "check_private_repo_access") as mock_priv_check, + ): + mock_run.side_effect = self._mock_subprocess_for_remote("fetch") + + response = client.post( + "/api/v1/git/fetch", + headers=session_auth_headers, + data=json.dumps({ + "repo_path": "/home/egg/repos/test", + "remote": "origin", + }), + content_type="application/json", + ) + + assert response.status_code == 200 + mock_priv_check.assert_not_called() + + def test_ls_remote_checkpoint_repo_bypasses_private_mode(self, client, session_auth_headers): + """ls-remote on a checkpoint repo skips check_private_repo_access.""" + with ( + patch("subprocess.run") as mock_run, + patch.object(gateway, "get_token_for_repo", return_value=("token", "bot", "")), + patch.object(gateway, "is_checkpoint_repo", return_value=True), + patch.object(gateway, "check_private_repo_access") as mock_priv_check, + ): + mock_run.side_effect = self._mock_subprocess_for_remote("ls-remote") + + response = client.post( + "/api/v1/git/fetch", + headers=session_auth_headers, + data=json.dumps({ + "repo_path": "/home/egg/repos/test", + "remote": "origin", + "operation": "ls-remote", + }), + content_type="application/json", + ) + + assert response.status_code == 200 + mock_priv_check.assert_not_called() + + def test_fetch_non_checkpoint_repo_still_checked(self, client, session_auth_headers): + """Fetch on a non-checkpoint repo still calls check_private_repo_access.""" + from private_repo_policy import PrivateRepoPolicyResult + + mock_policy_result = PrivateRepoPolicyResult( + allowed=True, + reason="Public repo allowed", + visibility="public", + ) + + with ( + patch("subprocess.run") as mock_run, + patch.object(gateway, "get_token_for_repo", return_value=("token", "bot", "")), + patch.object(gateway, "is_checkpoint_repo", return_value=False), + patch.object( + gateway, "check_private_repo_access", return_value=mock_policy_result + ) as mock_priv_check, + ): + mock_run.side_effect = self._mock_subprocess_for_remote("fetch") + + response = client.post( + "/api/v1/git/fetch", + headers=session_auth_headers, + data=json.dumps({ + "repo_path": "/home/egg/repos/test", + "remote": "origin", + }), + content_type="application/json", + ) + + assert response.status_code == 200 + mock_priv_check.assert_called_once() + + def test_push_checkpoint_repo_bypasses_private_mode(self, client, session_auth_headers): + """Push to a checkpoint repo skips check_private_repo_access.""" + with ( + patch("subprocess.run") as mock_run, + patch.object(gateway, "get_token_for_repo", return_value=("token", "bot", "")), + patch.object(gateway, "is_checkpoint_repo", return_value=True), + patch.object(gateway, "check_private_repo_access") as mock_priv_check, + patch.object(gateway, "get_policy_engine") as mock_policy, + ): + + def run_side_effect(*args, **kwargs): + cmd = args[0] if args else kwargs.get("args", []) + result = MagicMock() + result.returncode = 0 + result.stderr = "" + if "remote" in cmd and "get-url" in cmd: + result.stdout = "https://github.com/ckpt-owner/ckpt-repo.git\n" + elif "push" in cmd: + result.stdout = "" + else: + result.stdout = "" + return result + + mock_run.side_effect = run_side_effect + + # Policy engine allows push + mock_engine = MagicMock() + mock_engine.check_branch_ownership.return_value = PolicyResult( + allowed=True, reason="allowed" + ) + mock_policy.return_value = mock_engine + + response = client.post( + "/api/v1/git/push", + headers=session_auth_headers, + data=json.dumps({ + "repo_path": "/home/egg/repos/test", + "remote": "origin", + "refspec": "egg/checkpoints/v2", + }), + content_type="application/json", + ) + + assert response.status_code == 200 + mock_priv_check.assert_not_called() + + def test_push_checkpoint_branch_bypasses_private_mode(self, client, session_auth_headers): + """Push to egg/checkpoints/v2 branch skips check_private_repo_access even for non-checkpoint repos.""" + with ( + patch("subprocess.run") as mock_run, + patch.object(gateway, "get_token_for_repo", return_value=("token", "bot", "")), + patch.object(gateway, "is_checkpoint_repo", return_value=False), + patch.object(gateway, "check_private_repo_access") as mock_priv_check, + patch.object(gateway, "get_policy_engine") as mock_policy, + ): + + def run_side_effect(*args, **kwargs): + cmd = args[0] if args else kwargs.get("args", []) + result = MagicMock() + result.returncode = 0 + result.stderr = "" + if "remote" in cmd and "get-url" in cmd: + result.stdout = "https://github.com/owner/repo.git\n" + elif "push" in cmd: + result.stdout = "" + else: + result.stdout = "" + return result + + mock_run.side_effect = run_side_effect + + mock_engine = MagicMock() + mock_engine.check_branch_ownership.return_value = PolicyResult( + allowed=True, reason="allowed" + ) + mock_policy.return_value = mock_engine + + response = client.post( + "/api/v1/git/push", + headers=session_auth_headers, + data=json.dumps({ + "repo_path": "/home/egg/repos/test", + "remote": "origin", + "refspec": "egg/checkpoints/v2", + }), + content_type="application/json", + ) + + assert response.status_code == 200 + mock_priv_check.assert_not_called() + + def test_push_non_checkpoint_repo_still_checked(self, client, session_auth_headers): + """Push to a non-checkpoint repo still calls check_private_repo_access.""" + from private_repo_policy import PrivateRepoPolicyResult + + mock_policy_result = PrivateRepoPolicyResult( + allowed=True, + reason="Public repo allowed", + visibility="public", + ) + + with ( + patch("subprocess.run") as mock_run, + patch.object(gateway, "get_token_for_repo", return_value=("token", "bot", "")), + patch.object(gateway, "is_checkpoint_repo", return_value=False), + patch.object( + gateway, "check_private_repo_access", return_value=mock_policy_result + ) as mock_priv_check, + patch.object(gateway, "get_policy_engine") as mock_policy, + ): + + def run_side_effect(*args, **kwargs): + cmd = args[0] if args else kwargs.get("args", []) + result = MagicMock() + result.returncode = 0 + result.stderr = "" + if "remote" in cmd and "get-url" in cmd: + result.stdout = "https://github.com/owner/repo.git\n" + elif "push" in cmd: + result.stdout = "" + else: + result.stdout = "" + return result + + mock_run.side_effect = run_side_effect + + mock_engine = MagicMock() + mock_engine.check_branch_ownership.return_value = PolicyResult( + allowed=True, reason="allowed" + ) + mock_policy.return_value = mock_engine + + response = client.post( + "/api/v1/git/push", + headers=session_auth_headers, + data=json.dumps({ + "repo_path": "/home/egg/repos/test", + "remote": "origin", + "refspec": "egg/my-branch", + }), + content_type="application/json", + ) + + assert response.status_code == 200 + mock_priv_check.assert_called_once() diff --git a/tests/config/test_repo_config_checkpoint.py b/tests/config/test_repo_config_checkpoint.py new file mode 100644 index 0000000000..53a6e3ce4b --- /dev/null +++ b/tests/config/test_repo_config_checkpoint.py @@ -0,0 +1,203 @@ +""" +Tests for checkpoint repo helper functions in repo_config module. + +Tests get_all_checkpoint_repos() and is_checkpoint_repo() which are used +by the gateway to exempt checkpoint repos from private mode policy. +""" + +import pytest + +import config.repo_config as repo_config_module +from config.repo_config import get_all_checkpoint_repos, is_checkpoint_repo + + +@pytest.fixture(autouse=True) +def clear_checkpoint_cache(): + """Clear the checkpoint repos cache before each test.""" + repo_config_module._checkpoint_repos_cache = None + yield + repo_config_module._checkpoint_repos_cache = None + + +class TestGetAllCheckpointRepos: + """Tests for get_all_checkpoint_repos function.""" + + def test_empty_config(self, temp_dir, monkeypatch): + """Returns empty set when no repo_settings configured.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text("github_username: testuser\n") + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert result == set() + + def test_no_checkpoint_repos(self, temp_dir, monkeypatch): + """Returns empty set when repo_settings exist but no checkpoint_repo.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + " restrict_to_configured_users: true\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert result == set() + + def test_single_checkpoint_repo(self, temp_dir, monkeypatch): + """Returns set with one checkpoint repo.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + " checkpoint_repo: testuser/my-checkpoints\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert result == {"testuser/my-checkpoints"} + + def test_multiple_repos(self, temp_dir, monkeypatch): + """Returns set with multiple distinct checkpoint repos.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/app-one:\n" + " checkpoint_repo: testuser/ckpt-one\n" + " testuser/app-two:\n" + " checkpoint_repo: testuser/ckpt-two\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert result == {"testuser/ckpt-one", "testuser/ckpt-two"} + + def test_deduplication(self, temp_dir, monkeypatch): + """Multiple repos pointing to the same checkpoint repo are deduplicated.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/app-one:\n" + " checkpoint_repo: testuser/shared-ckpt\n" + " testuser/app-two:\n" + " checkpoint_repo: testuser/shared-ckpt\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert result == {"testuser/shared-ckpt"} + + def test_case_insensitivity(self, temp_dir, monkeypatch): + """Checkpoint repo names are lowercased for comparison.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + " checkpoint_repo: TestUser/My-Checkpoints\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert "testuser/my-checkpoints" in result + + def test_config_unavailable_returns_empty(self, temp_dir, monkeypatch): + """Returns empty set when config file cannot be loaded.""" + monkeypatch.setenv("EGG_REPO_CONFIG", str(temp_dir / "nonexistent.yaml")) + monkeypatch.setenv("HOME", str(temp_dir)) + + result = get_all_checkpoint_repos() + assert result == set() + + def test_ignores_non_string_checkpoint_repo(self, temp_dir, monkeypatch): + """Ignores checkpoint_repo values that are not strings.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + " checkpoint_repo: 12345\n" + " testuser/other-app:\n" + " checkpoint_repo: valid/repo\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert result == {"valid/repo"} + + def test_ignores_empty_checkpoint_repo(self, temp_dir, monkeypatch): + """Ignores empty string checkpoint_repo values.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + ' checkpoint_repo: ""\n' + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + result = get_all_checkpoint_repos() + assert result == set() + + +class TestIsCheckpointRepo: + """Tests for is_checkpoint_repo function.""" + + def test_match(self, temp_dir, monkeypatch): + """Returns True for a configured checkpoint repo.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + " checkpoint_repo: testuser/my-checkpoints\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + assert is_checkpoint_repo("testuser", "my-checkpoints") is True + + def test_no_match(self, temp_dir, monkeypatch): + """Returns False for a repo that is not a checkpoint destination.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + " checkpoint_repo: testuser/my-checkpoints\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + assert is_checkpoint_repo("testuser", "my-app") is False + + def test_case_insensitive(self, temp_dir, monkeypatch): + """Matching is case insensitive.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text( + "github_username: testuser\n" + "repo_settings:\n" + " testuser/my-app:\n" + " checkpoint_repo: TestUser/My-Checkpoints\n" + ) + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + assert is_checkpoint_repo("testuser", "my-checkpoints") is True + assert is_checkpoint_repo("TESTUSER", "MY-CHECKPOINTS") is True + + def test_config_unavailable_returns_false(self, temp_dir, monkeypatch): + """Returns False when config cannot be loaded (fail-closed).""" + monkeypatch.setenv("EGG_REPO_CONFIG", str(temp_dir / "nonexistent.yaml")) + monkeypatch.setenv("HOME", str(temp_dir)) + + assert is_checkpoint_repo("testuser", "my-checkpoints") is False + + def test_empty_config_returns_false(self, temp_dir, monkeypatch): + """Returns False when config has no checkpoint repos.""" + config_file = temp_dir / "repositories.yaml" + config_file.write_text("github_username: testuser\n") + monkeypatch.setenv("EGG_REPO_CONFIG", str(config_file)) + + assert is_checkpoint_repo("testuser", "my-checkpoints") is False