GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

Enforce deployment security boundaries

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 12:10:05 +02:00
parent 825791e9b0
commit 230ecf42b0
10 changed files with 629 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
tests/test_api_smoke.py
View File

@@ -1811,12 +1811,21 @@ class ApiSmokeTests(unittest.TestCase):
tenant_id = str(login["tenant"]["id"])
env_keys = [
"GOVOPLAN_FILES_CONNECTOR_PROFILES_JSON",
"GOVOPLAN_CONNECTOR_SECRET_ENV_ALLOWLIST",
"GOVOPLAN_TEST_SEAFILE_PASSWORD",
"GOVOPLAN_TEST_NEXTCLOUD_TOKEN",
"GOVOPLAN_TEST_SMB_PASSWORD",
]
previous_env = {key: os.environ.get(key) for key in env_keys}
try:
os.environ["GOVOPLAN_CONNECTOR_SECRET_ENV_ALLOWLIST"] = ",".join(
(
"GOVOPLAN_TEST_SEAFILE_PASSWORD",
"GOVOPLAN_TEST_WEBDAV_PASSWORD",
"GOVOPLAN_TEST_NEXTCLOUD_TOKEN",
"GOVOPLAN_TEST_SMB_PASSWORD",
)
)
os.environ["GOVOPLAN_TEST_SEAFILE_PASSWORD"] = "super-secret-seafile"
os.environ["GOVOPLAN_TEST_NEXTCLOUD_TOKEN"] = "super-secret-nextcloud"
os.environ["GOVOPLAN_TEST_SMB_PASSWORD"] = "super-secret-smb"
@@ -2203,36 +2212,37 @@ class ApiSmokeTests(unittest.TestCase):
self.assertFalse(deleted_space.json()["is_active"])
self.assertIsNotNone(deleted_space.json()["deleted_at"])
import httpx
from govoplan_files.backend.storage.http_client import ConnectorHttpResponse
def seafile_request(method: str, url: str, **kwargs):
request = httpx.Request(method, url)
params = kwargs.get("params") or {}
if url == "http://127.0.0.1:9082/api2/auth-token/":
return httpx.Response(200, json={"token": "token-1"}, request=request)
return ConnectorHttpResponse(200, {}, b'{"token":"token-1"}')
if url == "http://127.0.0.1:9082/api2/repos/repo-1/file/detail/":
self.assertEqual(params.get("p"), "/reports/summary.txt")
self.assertEqual(kwargs.get("headers", {}).get("Authorization"), "Token token-1")
return httpx.Response(
return ConnectorHttpResponse(
200,
json={
"id": "file-1",
"name": "summary.txt",
"size": 14,
"type": "file",
"mtime": 1783425600,
"permission": "r",
},
request=request,
{},
json.dumps(
{
"id": "file-1",
"name": "summary.txt",
"size": 14,
"type": "file",
"mtime": 1783425600,
"permission": "r",
}
).encode(),
)
if url == "http://127.0.0.1:9082/api2/repos/repo-1/file/":
self.assertEqual(params, {"p": "/reports/summary.txt", "reuse": "1"})
return httpx.Response(200, json="https://download.example.invalid/summary.txt", request=request)
return ConnectorHttpResponse(200, {}, b'"https://download.example.invalid/summary.txt"')
if url == "https://download.example.invalid/summary.txt":
return httpx.Response(200, content=b"summary report", headers={"content-type": "text/plain"}, request=request)
return httpx.Response(404, request=request)
return ConnectorHttpResponse(200, {"content-type": "text/plain"}, b"summary report")
return ConnectorHttpResponse(404, {}, b"")
with patch("govoplan_files.backend.storage.connector_browse.httpx.request", side_effect=seafile_request), patch("govoplan_files.backend.storage.connector_imports.httpx.request", side_effect=seafile_request):
with patch("govoplan_files.backend.storage.connector_browse.request_connector_bytes", side_effect=seafile_request), patch("govoplan_files.backend.storage.connector_imports.request_connector_bytes", side_effect=seafile_request):
imported = self.client.post(
"/api/v1/files/connectors/profiles/seafile-dev/import",
headers=headers,
@@ -2259,18 +2269,16 @@ class ApiSmokeTests(unittest.TestCase):
webdav_payload = {"content": b"nextcloud notice", "etag": "\"nextcloud-etag-2\""}
def webdav_request(method: str, url: str, **kwargs):
request = httpx.Request(method, url)
self.assertEqual(method, "GET")
self.assertEqual(url, "http://127.0.0.1:9081/Shared/notice.txt")
self.assertEqual(kwargs.get("headers", {}).get("Authorization"), "Bearer super-secret-nextcloud")
return httpx.Response(
return ConnectorHttpResponse(
200,
content=webdav_payload["content"],
headers={"content-type": "text/plain", "etag": webdav_payload["etag"], "content-length": str(len(webdav_payload["content"]))},
request=request,
{"content-type": "text/plain", "etag": webdav_payload["etag"], "content-length": str(len(webdav_payload["content"]))},
webdav_payload["content"],
)
with patch("govoplan_files.backend.storage.connector_imports.httpx.request", side_effect=webdav_request):
with patch("govoplan_files.backend.storage.connector_imports.request_connector_bytes", side_effect=webdav_request):
nextcloud_import = self.client.post(
"/api/v1/files/connectors/profiles/user-nextcloud/import",
headers=headers,
@@ -2292,7 +2300,7 @@ class ApiSmokeTests(unittest.TestCase):
webdav_payload["content"] = b"nextcloud notice updated"
webdav_payload["etag"] = "\"nextcloud-etag-3\""
with patch("govoplan_files.backend.storage.connector_imports.httpx.request", side_effect=webdav_request):
with patch("govoplan_files.backend.storage.connector_imports.request_connector_bytes", side_effect=webdav_request):
nextcloud_sync = self.client.post(
"/api/v1/files/connectors/profiles/user-nextcloud/sync",
headers=headers,
@@ -2314,7 +2322,7 @@ class ApiSmokeTests(unittest.TestCase):
self.assertEqual(synced["file"]["source_revision"], "\"nextcloud-etag-3\"")
self.assertEqual(synced["file"]["size_bytes"], len(webdav_payload["content"]))
with patch("govoplan_files.backend.storage.connector_imports.httpx.request", side_effect=webdav_request):
with patch("govoplan_files.backend.storage.connector_imports.request_connector_bytes", side_effect=webdav_request):
nextcloud_sync_unchanged = self.client.post(
"/api/v1/files/connectors/profiles/user-nextcloud/sync",
headers=headers,