fix(proxy): add domain validation for proxy requests (#1740)

gustavobtflores · web-flow · commit 54eab00aabd5 · 2026-02-10T19:29:11.000Z
* fix(proxy): add domain validation for proxy requests

* test(proxy): enhance proxy view tests with domain allowlist

- Added tests for allowed and forbidden domains, including wildcard and exact matches.
- Updated existing tests to use new allowed domains and paths for consistency.

* fix(logViewer): handle domain not allowed error
diff --git a/backend/kernelCI_app/constants/localization.py b/backend/kernelCI_app/constants/localization.py
@@ -24,6 +24,7 @@ class ClientStrings:
     NO_TREES_FOUND = "No trees were found"
     PROXY_FETCH_FAILED = "Failed to fetch resource:"
     PROXY_INVALID_URL = "Invalid URL"
+    PROXY_FORBIDDEN_DOMAIN = "Domain not allowed"
     PROXY_ERROR_FETCH = "Error fetching the resource"
     NO_ORIGIN_FOUND = "No origins found"
     LOG_TABLE_NOT_FOUND = "Table with id 'list' not found"
diff --git a/backend/kernelCI_app/tests/unitTests/views/proxyView_test.py b/backend/kernelCI_app/tests/unitTests/views/proxyView_test.py
@@ -8,13 +8,17 @@
 from kernelCI_app.constants.localization import ClientStrings
 from kernelCI_app.views.proxyView import TIMEOUT_TIME_IN_SECONDS, ProxyView
 
+MOCK_ALLOWED_DOMAINS = ["*.example.com", "exact.org"]
+MOCK_ALLOWED_S3_PATHS = ["/allowed-bucket/"]
+
 
 class TestProxyView(SimpleTestCase):
     def setUp(self):
         self.factory = APIRequestFactory()
         self.view = ProxyView()
         self.url = "/proxy/"
 
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", MOCK_ALLOWED_DOMAINS)
     @patch("kernelCI_app.views.proxyView.requests.get")
     def test_get_proxy_success(self, mock_requests_get):
         mock_requests_get.return_value.status_code = HTTPStatus.OK
@@ -25,7 +29,9 @@ def test_get_proxy_success(self, mock_requests_get):
             "Content-Disposition": "attachment; filename=file.txt",
         }
 
-        request = self.factory.get(self.url, {"url": "https://example.com/file.txt"})
+        request = self.factory.get(
+            self.url, {"url": "https://files.example.com/file.txt"}
+        )
         response = self.view.get(request)
 
         self.assertEqual(response.status_code, 200)
@@ -36,16 +42,21 @@ def test_get_proxy_success(self, mock_requests_get):
             response["Content-Disposition"], "attachment; filename=file.txt"
         )
         mock_requests_get.assert_called_once_with(
-            "https://example.com/file.txt", stream=True, timeout=TIMEOUT_TIME_IN_SECONDS
+            "https://files.example.com/file.txt",
+            stream=True,
+            timeout=TIMEOUT_TIME_IN_SECONDS,
         )
 
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", MOCK_ALLOWED_DOMAINS)
     @patch("kernelCI_app.views.proxyView.requests.get")
     def test_get_proxy_default_content_type(self, mock_requests_get):
         mock_requests_get.return_value.status_code = HTTPStatus.OK
         mock_requests_get.return_value.content = b"content"
         mock_requests_get.return_value.headers = {}
 
-        request = self.factory.get(self.url, {"url": "https://example.com/file.txt"})
+        request = self.factory.get(
+            self.url, {"url": "https://files.example.com/file.txt"}
+        )
         response = self.view.get(request)
 
         self.assertEqual(response.status_code, 200)
@@ -65,23 +76,98 @@ def test_get_proxy_missing_url(self):
         self.assertEqual(response.status_code, HTTPStatus.BAD_REQUEST)
         self.assertEqual(response.data, {"error": ClientStrings.PROXY_INVALID_URL})
 
+    def test_get_proxy_invalid_scheme(self):
+        request = self.factory.get(self.url, {"url": "file:///etc/passwd"})
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.BAD_REQUEST)
+        self.assertEqual(response.data, {"error": ClientStrings.PROXY_INVALID_URL})
+
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", MOCK_ALLOWED_DOMAINS)
     @patch("kernelCI_app.views.proxyView.requests.get")
     def test_get_proxy_external_error(self, mock_requests_get):
         mock_requests_get.return_value.status_code = HTTPStatus.NOT_FOUND
         mock_requests_get.return_value.reason = "Not Found"
 
-        request = self.factory.get(self.url, {"url": "https://example.com/file.txt"})
+        request = self.factory.get(
+            self.url, {"url": "https://files.example.com/file.txt"}
+        )
         response = self.view.get(request)
 
         self.assertEqual(response.status_code, HTTPStatus.NOT_FOUND)
         self.assertIn(ClientStrings.PROXY_FETCH_FAILED, response.data["error"])
 
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", MOCK_ALLOWED_DOMAINS)
     @patch("kernelCI_app.views.proxyView.requests.get")
     def test_get_proxy_request_exception(self, mock_requests_get):
         mock_requests_get.side_effect = requests.RequestException("Connection error")
 
-        request = self.factory.get(self.url, {"url": "https://example.com/file.txt"})
+        request = self.factory.get(
+            self.url, {"url": "https://files.example.com/file.txt"}
+        )
         response = self.view.get(request)
 
         self.assertEqual(response.status_code, HTTPStatus.INTERNAL_SERVER_ERROR)
         self.assertEqual(response.data, {"error": ClientStrings.PROXY_ERROR_FETCH})
+
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", MOCK_ALLOWED_DOMAINS)
+    def test_get_proxy_forbidden_domain(self):
+        request = self.factory.get(self.url, {"url": "https://evil.com/file.txt"})
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.FORBIDDEN)
+        self.assertEqual(response.data, {"error": ClientStrings.PROXY_FORBIDDEN_DOMAIN})
+
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", MOCK_ALLOWED_DOMAINS)
+    @patch("kernelCI_app.views.proxyView.requests.get")
+    def test_get_proxy_wildcard_domain_match(self, mock_requests_get):
+        mock_requests_get.return_value.status_code = HTTPStatus.OK
+        mock_requests_get.return_value.content = b"content"
+        mock_requests_get.return_value.headers = {}
+
+        request = self.factory.get(
+            self.url, {"url": "https://any-sub.example.com/file.txt"}
+        )
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, 200)
+
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", MOCK_ALLOWED_DOMAINS)
+    @patch("kernelCI_app.views.proxyView.requests.get")
+    def test_get_proxy_exact_domain_match(self, mock_requests_get):
+        mock_requests_get.return_value.status_code = HTTPStatus.OK
+        mock_requests_get.return_value.content = b"content"
+        mock_requests_get.return_value.headers = {}
+
+        request = self.factory.get(self.url, {"url": "https://exact.org/file.txt"})
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, 200)
+
+    @patch("kernelCI_app.views.proxyView.ALLOWED_S3_PATHS", MOCK_ALLOWED_S3_PATHS)
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", ["s3.amazonaws.com"])
+    @patch("kernelCI_app.views.proxyView.requests.get")
+    def test_get_proxy_s3_allowed_bucket(self, mock_requests_get):
+        mock_requests_get.return_value.status_code = HTTPStatus.OK
+        mock_requests_get.return_value.content = b"log content"
+        mock_requests_get.return_value.headers = {}
+
+        request = self.factory.get(
+            self.url,
+            {"url": "https://s3.amazonaws.com/allowed-bucket/file.log"},
+        )
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, 200)
+
+    @patch("kernelCI_app.views.proxyView.ALLOWED_S3_PATHS", MOCK_ALLOWED_S3_PATHS)
+    @patch("kernelCI_app.views.proxyView.ALLOWED_DOMAINS", ["s3.amazonaws.com"])
+    def test_get_proxy_s3_forbidden_bucket(self):
+        request = self.factory.get(
+            self.url,
+            {"url": "https://s3.amazonaws.com/other-bucket/file.log"},
+        )
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.FORBIDDEN)
+        self.assertEqual(response.data, {"error": ClientStrings.PROXY_FORBIDDEN_DOMAIN})
diff --git a/backend/kernelCI_app/views/proxyView.py b/backend/kernelCI_app/views/proxyView.py
@@ -1,5 +1,6 @@
-from urllib.parse import urlparse
+from urllib.parse import ParseResult, urlparse
 import requests
+from fnmatch import fnmatch
 from rest_framework.views import APIView
 from django.http import HttpResponse
 from http import HTTPStatus
@@ -11,6 +12,46 @@
 # TIMEOUT to avoid people sending very large files through the proxy
 TIMEOUT_TIME_IN_SECONDS = 45
 
+ALLOWED_DOMAINS = [
+    # KernelCI infrastructure
+    "*.kernelci.org",
+    # External CI/test infrastructure
+    "syzkaller.appspot.com",
+    "*.sirena.org.uk",
+    "lisalogs6a28e896.blob.core.windows.net",
+    "storage.tuxsuite.com",
+    "s3.amazonaws.com",
+    "*.linaro.org",
+]
+
+ALLOWED_S3_PATHS = [
+    "/arr-cki-prod-trusted-artifacts/",
+]
+
+
+def is_valid_url(parsed_url: ParseResult) -> bool:
+    if not parsed_url.scheme or not parsed_url.netloc:
+        return False
+
+    if parsed_url.scheme not in ["http", "https"]:
+        return False
+
+    return True
+
+
+def is_allowed_domain(parsed_url: ParseResult) -> bool:
+    if not any(fnmatch(parsed_url.hostname, pattern) for pattern in ALLOWED_DOMAINS):
+        return False
+
+    if parsed_url.hostname == "s3.amazonaws.com":
+        if not any(
+            parsed_url.path.startswith(allowed_path)
+            for allowed_path in ALLOWED_S3_PATHS
+        ):
+            return False
+
+    return True
+
 
 class ProxyView(APIView):
     @extend_schema(
@@ -20,11 +61,24 @@ class ProxyView(APIView):
     )
     def get(self, request):
         url = request.GET.get("url")
+        if not url:
+            return create_api_error_response(
+                error_message=ClientStrings.PROXY_INVALID_URL
+            )
+
         parsed_url = urlparse(url)
-        if not parsed_url.scheme or not parsed_url.netloc:
+
+        if not is_valid_url(parsed_url):
             return create_api_error_response(
                 error_message=ClientStrings.PROXY_INVALID_URL
             )
+
+        if not is_allowed_domain(parsed_url):
+            return create_api_error_response(
+                error_message=ClientStrings.PROXY_FORBIDDEN_DOMAIN,
+                status_code=HTTPStatus.FORBIDDEN,
+            )
+
         try:
             response = requests.get(url, stream=True, timeout=TIMEOUT_TIME_IN_SECONDS)
 
diff --git a/dashboard/src/api/logViewer.ts b/dashboard/src/api/logViewer.ts
@@ -4,6 +4,8 @@ import { useQuery } from '@tanstack/react-query';
 import { minutesToMilliseconds } from 'date-fns';
 import { useIntl } from 'react-intl';
 
+import { AxiosError, HttpStatusCode } from 'axios';
+
 import { LOG_EXCERPT_ALLOWED_DOMAINS } from '@/utils/constants/log_excerpt_allowed_domain';
 
 import { RequestData } from './commonRequest';
@@ -43,6 +45,14 @@ async function fetchAndDecompressLog(
     }
   } catch (error) {
     console.error(error);
+
+    if (
+      error instanceof AxiosError &&
+      error.response?.status === HttpStatusCode.Forbidden
+    ) {
+      throw new Error('403:Domain not allowed');
+    }
+
     throw new Error(
       `Failed to fetch logs: ${error instanceof Error ? error.message : 'Unknown error'}`,
     );
diff --git a/dashboard/src/locales/messages/index.ts b/dashboard/src/locales/messages/index.ts
@@ -255,6 +255,8 @@ export const messages = {
     'logSheet.title': 'Log Viewer',
     'logViewer.disabledHighlight':
       'The text highlight was disabled in this log due to the log size',
+    'logViewer.domainNotAllowed':
+      'This log domain is not supported by the proxy. You can still download the log directly using the link above.',
     'logViewer.download': 'Download full log',
     'logViewer.errorFetchingLogExcerpt':
       'Error fetching logExcerpt at: {logExcerpt}',
diff --git a/dashboard/src/pages/LogViewer.tsx b/dashboard/src/pages/LogViewer.tsx
@@ -6,7 +6,7 @@ const Constants = {
   URL_END_PATH_SIZE: 50,
 } as const;
 
-import { FormattedMessage } from 'react-intl';
+import { FormattedMessage, useIntl } from 'react-intl';
 
 import { LogExcerpt } from '@/components/Log/LogExcerpt';
 import { truncateUrl } from '@/lib/string';
@@ -57,8 +57,16 @@ export function LogViewer(): JSX.Element {
   const { url, type, itemId } = useSearch({ from: '/log-viewer' });
   const historyState = useRouterState({ select: s => s.location.state });
   const previousSearch = useSearchStore(s => s.previousSearch);
+  const { formatMessage } = useIntl();
 
-  const { data, status } = useLogViewer(url);
+  const { data, status, error } = useLogViewer(url);
+
+  const isDomainNotAllowed = error?.message.startsWith('403');
+  const domainNotAllowedError = isDomainNotAllowed ? (
+    <p className="text-weak-gray py-6 text-center text-xl font-semibold">
+      {formatMessage({ id: 'logViewer.domainNotAllowed' })}
+    </p>
+  ) : undefined;
   const { data: logData, isLoading } = useLogData(itemId ?? '', type);
 
   const { treeName, branch, id } = historyState;
@@ -192,7 +200,12 @@ export function LogViewer(): JSX.Element {
         </div>
       </div>
       <div className="overflow-hidden rounded-md">
-        <QuerySwitcher data={data} status={status}>
+        <QuerySwitcher
+          data={data}
+          status={status}
+          error={error}
+          customError={domainNotAllowedError}
+        >
           <LogExcerpt logExcerpt={data?.content} variant="log-viewer" />
         </QuerySwitcher>
       </div>
