diff --git a/backend/app/routes.py b/backend/app/routes.py index 1745f89..7c76ee8 100644 --- a/backend/app/routes.py +++ b/backend/app/routes.py @@ -7866,7 +7866,6 @@ from .upstream import ( UpstreamTimeoutError, UpstreamHTTPError, UpstreamSSLError, - AirGapError, FileSizeExceededError as UpstreamFileSizeExceededError, SourceNotFoundError, SourceDisabledError, @@ -8114,8 +8113,6 @@ def cache_artifact( cache_request.url, expected_hash=cache_request.expected_hash, ) - except AirGapError as e: - raise HTTPException(status_code=403, detail=str(e)) except SourceDisabledError as e: raise HTTPException(status_code=503, detail=str(e)) except UpstreamHTTPError as e: diff --git a/backend/app/upstream.py b/backend/app/upstream.py index 4e8995f..5856c05 100644 --- a/backend/app/upstream.py +++ b/backend/app/upstream.py @@ -278,7 +278,6 @@ class UpstreamClient: FetchResult with content, hash, size, and headers. Raises: - AirGapError: If air-gap mode blocks the request. SourceDisabledError: If the matching source is disabled. UpstreamConnectionError: On connection failures. UpstreamTimeoutError: On timeout. diff --git a/backend/tests/test_upstream_caching.py b/backend/tests/test_upstream_caching.py index 84917bf..49b105d 100644 --- a/backend/tests/test_upstream_caching.py +++ b/backend/tests/test_upstream_caching.py @@ -91,7 +91,6 @@ class TestUpstreamSourceModel: assert hasattr(source, 'source_type') assert hasattr(source, 'url') assert hasattr(source, 'enabled') - assert hasattr(source, 'is_public') assert hasattr(source, 'auth_type') assert hasattr(source, 'username') assert hasattr(source, 'password_encrypted') @@ -107,7 +106,6 @@ class TestUpstreamSourceModel: source_type="npm", url="https://npm.example.com", enabled=True, - is_public=False, auth_type="basic", username="admin", priority=50, @@ -116,7 +114,6 @@ class TestUpstreamSourceModel: assert source.source_type == "npm" assert source.url == "https://npm.example.com" assert source.enabled is True - assert source.is_public is False assert source.auth_type == "basic" assert source.username == "admin" assert source.priority == 50 @@ -260,7 +257,6 @@ class TestUpstreamSourceSchemas: source_type="npm", url="https://npm.example.com", enabled=True, - is_public=False, auth_type="basic", username="admin", password="secret", @@ -281,7 +277,6 @@ class TestUpstreamSourceSchemas: ) assert source.source_type == "generic" assert source.enabled is False - assert source.is_public is True assert source.auth_type == "none" assert source.priority == 100 @@ -578,7 +573,6 @@ class TestUpstreamClientSourceMatching: name="npm-public", url="https://registry.npmjs.org", enabled=True, - is_public=True, auth_type="none", priority=100, ) @@ -603,7 +597,6 @@ class TestUpstreamClientSourceMatching: name="npm-private", url="https://registry.npmjs.org", enabled=True, - is_public=False, auth_type="basic", priority=50, ) @@ -611,7 +604,6 @@ class TestUpstreamClientSourceMatching: name="npm-public", url="https://registry.npmjs.org", enabled=True, - is_public=True, auth_type="none", priority=100, ) @@ -711,89 +703,6 @@ class TestUpstreamClientAuthHeaders: assert auth is None -class TestUpstreamClientAirGapMode: - """Tests for air-gap mode enforcement.""" - - def test_airgap_blocks_public_source(self): - """Test that air-gap mode blocks public sources.""" - from app.models import UpstreamSource, CacheSettings - from app.upstream import UpstreamClient, AirGapError - - source = UpstreamSource( - name="npm-public", - url="https://registry.npmjs.org", - enabled=True, - is_public=True, - auth_type="none", - priority=100, - ) - settings = CacheSettings(allow_public_internet=False) - - client = UpstreamClient(sources=[source], cache_settings=settings) - - with pytest.raises(AirGapError) as exc_info: - client.fetch("https://registry.npmjs.org/lodash") - - assert "Air-gap mode enabled" in str(exc_info.value) - assert "public source" in str(exc_info.value) - - def test_airgap_blocks_unmatched_url(self): - """Test that air-gap mode blocks URLs not matching any source.""" - from app.models import CacheSettings - from app.upstream import UpstreamClient, AirGapError - - settings = CacheSettings(allow_public_internet=False) - client = UpstreamClient(sources=[], cache_settings=settings) - - with pytest.raises(AirGapError) as exc_info: - client.fetch("https://example.com/file.tgz") - - assert "Air-gap mode enabled" in str(exc_info.value) - assert "does not match any configured" in str(exc_info.value) - - def test_airgap_allows_private_source(self): - """Test that air-gap mode allows private sources.""" - from app.models import UpstreamSource, CacheSettings - from app.upstream import UpstreamClient, SourceDisabledError - - source = UpstreamSource( - name="npm-private", - url="https://npm.internal.corp", - enabled=False, # Disabled, but would pass air-gap check - is_public=False, - auth_type="none", - priority=100, - ) - settings = CacheSettings(allow_public_internet=False) - - client = UpstreamClient(sources=[source], cache_settings=settings) - - # Should fail due to disabled source, not air-gap - with pytest.raises(SourceDisabledError): - client.fetch("https://npm.internal.corp/package.tgz") - - def test_allow_public_internet_true(self): - """Test that public internet is allowed when setting is true.""" - from app.models import UpstreamSource, CacheSettings - from app.upstream import UpstreamClient, SourceDisabledError - - source = UpstreamSource( - name="npm-public", - url="https://registry.npmjs.org", - enabled=False, # Disabled - is_public=True, - auth_type="none", - priority=100, - ) - settings = CacheSettings(allow_public_internet=True) - - client = UpstreamClient(sources=[source], cache_settings=settings) - - # Should fail due to disabled source, not air-gap - with pytest.raises(SourceDisabledError): - client.fetch("https://registry.npmjs.org/lodash") - - class TestUpstreamClientSourceDisabled: """Tests for disabled source handling.""" @@ -806,7 +715,6 @@ class TestUpstreamClientSourceDisabled: name="npm-public", url="https://registry.npmjs.org", enabled=False, - is_public=True, auth_type="none", priority=100, ) @@ -979,13 +887,6 @@ class TestUpstreamExceptions: assert error.status_code == 404 assert error.response_headers == {"x-custom": "value"} - def test_airgap_error(self): - """Test AirGapError.""" - from app.upstream import AirGapError - - error = AirGapError("Blocked by air-gap") - assert "Blocked by air-gap" in str(error) - def test_source_not_found_error(self): """Test SourceNotFoundError.""" from app.upstream import SourceNotFoundError @@ -1420,7 +1321,6 @@ class TestUpstreamSourcesAdminAPI: "source_type": "generic", "url": "https://example.com/packages", "enabled": False, - "is_public": False, "auth_type": "none", "priority": 200, }, @@ -1432,7 +1332,6 @@ class TestUpstreamSourcesAdminAPI: assert data["source_type"] == "generic" assert data["url"] == "https://example.com/packages" assert data["enabled"] is False - assert data["is_public"] is False assert data["priority"] == 200 assert "id" in data @@ -1452,7 +1351,6 @@ class TestUpstreamSourcesAdminAPI: "source_type": "npm", "url": "https://npm.internal.corp", "enabled": False, - "is_public": False, "auth_type": "basic", "username": "reader", "password": "secret123", @@ -1958,7 +1856,6 @@ class TestEnvVarUpstreamSourcesParsing: # Check defaults assert test_source.source_type == "generic" assert test_source.enabled is True - assert test_source.is_public is True assert test_source.auth_type == "none" assert test_source.priority == 100 finally: @@ -1981,7 +1878,6 @@ class TestEnvSourceToResponse: url="https://example.com", source_type="npm", enabled=True, - is_public=False, auth_type="basic", username="user", password="pass", @@ -1992,7 +1888,6 @@ class TestEnvSourceToResponse: assert source.url == "https://example.com" assert source.source_type == "npm" assert source.enabled is True - assert source.is_public is False assert source.auth_type == "basic" assert source.username == "user" assert source.password == "pass"