Apply audit remediation

Author: Pigbibi

.github/workflows/monthly_snapshot_audit.yml

@@ -5,10 +5,15 @@ name: Monthly Snapshot Audit
     - cron: "25 2 1 * *"
   workflow_dispatch:
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: false
+
 jobs:
   monthly-snapshot-audit:
     if: github.ref_name != 'logs'
     runs-on:
+    timeout-minutes: 60
       - self-hosted
       - Linux
       - X64

.github/workflows/publish-hk-snapshot-artifacts.yml

@@ -95,9 +95,14 @@ on:
         default: longbridgequant
         type: string
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: false
+
 jobs:
   build-and-publish:
     runs-on: ubuntu-latest
+    timeout-minutes: 60
     permissions:
       contents: read
       id-token: write

src/hk_equity_snapshot_pipelines/input_sources.py

@@ -1,17 +1,19 @@
 from __future__ import annotations
 
 import argparse
+import shutil
 import shlex
 import subprocess
 from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from urllib.parse import urlparse
-from urllib.request import urlretrieve
+from urllib.request import Request, urlopen
 
 from .evidence_uri_policy import SENSITIVE_EVIDENCE_URI_MARKERS
 
 SUPPORTED_SNAPSHOT_SUFFIXES = frozenset({".csv"})
+DEFAULT_REMOTE_COPY_TIMEOUT_SECONDS = 60
 
 CopyFn = Callable[[str, Path], None]
 
@@ -47,7 +49,9 @@ def _default_gcs_copy(source: str, target: Path) -> None:
 
 
 def _default_https_copy(source: str, target: Path) -> None:
-    urlretrieve(source, target)  # noqa: S310 - operator-supplied HTTPS data source URL.
+    request = Request(source, headers={"User-Agent": "hk-equity-snapshot-pipelines"})
+    with urlopen(request, timeout=DEFAULT_REMOTE_COPY_TIMEOUT_SECONDS) as response, target.open("wb") as output:  # noqa: S310 - operator-supplied HTTPS data source URL.
+        shutil.copyfileobj(response, output)
 
 
 def _source_suffix(source: str, *, allowed_suffixes: frozenset[str], default_suffix: str) -> str:

tests/test_input_sources.py

@@ -7,6 +7,8 @@
 import pytest
 
 from hk_equity_snapshot_pipelines.input_sources import (
+    DEFAULT_REMOTE_COPY_TIMEOUT_SECONDS,
+    _default_https_copy,
     resolve_hk_snapshot_inputs,
     resolve_input_source,
     source_needs_gcloud,
@@ -68,6 +70,36 @@ def fake_copy(source: str, target: Path) -> None:
     assert resolved.exists()
 
 
+def test_default_https_copy_uses_timeout_and_streams_response(monkeypatch, tmp_path):
+    calls: list[tuple[str, float]] = []
+
+    class FakeResponse:
+        def __init__(self) -> None:
+            self._payload = b"symbol,sector\n00001,Financials\n"
+
+        def __enter__(self) -> "FakeResponse":
+            return self
+
+        def __exit__(self, *args: object) -> None:
+            return None
+
+        def read(self, size: int = -1) -> bytes:
+            payload, self._payload = self._payload, b""
+            return payload
+
+    def fake_urlopen(request, *, timeout: float):
+        calls.append((request.full_url, timeout))
+        return FakeResponse()
+
+    monkeypatch.setattr("hk_equity_snapshot_pipelines.input_sources.urlopen", fake_urlopen)
+    target = tmp_path / "factor_snapshot.csv"
+
+    _default_https_copy("https://example.com/factor_snapshot.csv", target)
+
+    assert target.read_text(encoding="utf-8") == "symbol,sector\n00001,Financials\n"
+    assert calls == [("https://example.com/factor_snapshot.csv", DEFAULT_REMOTE_COPY_TIMEOUT_SECONDS)]
+
+
 def test_resolve_input_source_rejects_secret_like_remote_uri(tmp_path):
     with pytest.raises(ValueError, match="must not contain token"):
         resolve_input_source(