Wrap Infisical auth HTTPStatusError as ProviderError

psi/cli.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 from typing import Annotated
 
+import httpx
 import typer
 from pydantic import ValidationError
 from rich.console import Console
@@ -435,6 +436,9 @@ def main() -> None:
     except ValidationError as e:
         _print_validation_error(e)
         raise SystemExit(1) from e
+    except httpx.HTTPError as e:
+        _print_error(f"Network error: {e}")
+        raise SystemExit(1) from e
     except Exception as e:
         _print_bug()
         raise SystemExit(2) from e

psi/providers/infisical/auth.py

@@ -8,12 +8,11 @@
 
 import base64
 import json
-from typing import TYPE_CHECKING
 
-from psi.providers.infisical.models import AuthConfig, AuthMethod
+import httpx
 
-if TYPE_CHECKING:
-    import httpx
+from psi.errors import ProviderError
+from psi.providers.infisical.models import AuthConfig, AuthMethod
 
 # STS endpoint for AWS IAM auth — global endpoint works from any region
 _AWS_STS_ENDPOINT = "https://sts.amazonaws.com"
@@ -51,7 +50,12 @@ def authenticate(
 
 def _parse_token_response(response: httpx.Response) -> tuple[str, int]:
     """Extract access token and expiry from Infisical auth response."""
-    response.raise_for_status()
+    try:
+        response.raise_for_status()
+    except httpx.HTTPStatusError as e:
+        body = e.response.text[:200]
+        msg = f"Infisical authentication failed (HTTP {e.response.status_code}): {body}"
+        raise ProviderError(msg, provider_name="infisical") from e
     data = response.json()
     return data["accessToken"], int(data["expiresIn"])
 

tests/test_cli_error_handling.py

@@ -4,6 +4,7 @@
 
 from unittest.mock import patch
 
+import httpx
 import pytest
 
 from psi.errors import ConfigError, ProviderError, PsiError
@@ -59,3 +60,32 @@ def test_keyboard_interrupt_exits_130(self) -> None:
 
             with pytest.raises(SystemExit, match="130"):
                 main()
+
+    def test_httpx_connect_error_exits_1_clean(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """Stray httpx errors surface as a clean 'Network error', not a bug panic."""
+        with patch("psi.cli.app", side_effect=httpx.ConnectError("refused")):
+            from psi.cli import main
+
+            with pytest.raises(SystemExit, match="1"):
+                main()
+
+        captured = capsys.readouterr()
+        assert "Network error" in captured.out
+        assert "refused" in captured.out
+        assert "this is a bug" not in captured.out
+
+    def test_httpx_status_error_exits_1_clean(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """A raw HTTPStatusError (missed by provider wrapping) is still clean."""
+        request = httpx.Request("POST", "http://test")
+        response = httpx.Response(502, request=request)
+        exc = httpx.HTTPStatusError("502", request=request, response=response)
+
+        with patch("psi.cli.app", side_effect=exc):
+            from psi.cli import main
+
+            with pytest.raises(SystemExit, match="1"):
+                main()
+
+        captured = capsys.readouterr()
+        assert "Network error" in captured.out
+        assert "this is a bug" not in captured.out

tests/test_infisical_auth.py

@@ -0,0 +1,33 @@
+"""Tests for psi.providers.infisical.auth error wrapping."""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from psi.errors import ProviderError
+from psi.providers.infisical.auth import _parse_token_response
+
+
+def _response(status_code: int, body: bytes = b"") -> httpx.Response:
+    request = httpx.Request("POST", "http://test/api/v1/auth/universal-auth/login")
+    return httpx.Response(status_code, request=request, content=body)
+
+
+class TestParseTokenResponse:
+    def test_success_returns_token_and_ttl(self) -> None:
+        resp = _response(200, b'{"accessToken": "tok", "expiresIn": 3600}')
+        assert _parse_token_response(resp) == ("tok", 3600)
+
+    def test_502_raises_provider_error_with_cause(self) -> None:
+        resp = _response(502, b"Bad Gateway")
+        with pytest.raises(ProviderError) as excinfo:
+            _parse_token_response(resp)
+        assert "HTTP 502" in str(excinfo.value)
+        assert excinfo.value.provider_name == "infisical"
+        assert isinstance(excinfo.value.__cause__, httpx.HTTPStatusError)
+
+    def test_401_raises_provider_error_with_body_snippet(self) -> None:
+        resp = _response(401, b'{"error": "invalid credentials"}')
+        with pytest.raises(ProviderError, match="invalid credentials"):
+            _parse_token_response(resp)

tests/test_setup.py

@@ -8,10 +8,12 @@
 import httpx
 import pytest
 
+from psi.errors import ProviderError
 from psi.models import SecretSource, SystemdScope, WorkloadConfig
 from psi.providers.infisical import InfisicalProvider
 from psi.settings import PsiSettings
 from psi.setup import (
+    _RETRY_DELAYS,
     _generate_drop_in,
     _is_retryable,
     _setup_infisical_workload,
@@ -274,3 +276,70 @@ def test_non_retryable_error_raises_immediately(self, tmp_path: Path) -> None:
             pytest.raises(httpx.HTTPStatusError, match="unauthorized"),
         ):
             _setup_infisical_workload(settings, "myapp", {})
+
+    def test_auth_502_retries_then_raises_provider_error(self, tmp_path: Path) -> None:
+        """Auth endpoint 502 wrapped as ProviderError is retried via __cause__."""
+        request = httpx.Request("POST", "http://test/api/v1/auth/universal-auth/login")
+        response = httpx.Response(502, request=request)
+        call_count = 0
+
+        def mock_fetch(settings, workload_name, cache_updates):
+            nonlocal call_count
+            call_count += 1
+            http_err = httpx.HTTPStatusError("502", request=request, response=response)
+            raise ProviderError(
+                "Infisical authentication failed (HTTP 502): ...",
+                provider_name="infisical",
+            ) from http_err
+
+        settings = _make_settings(
+            tmp_path,
+            workloads={
+                "myapp": WorkloadConfig(
+                    provider="infisical",
+                    secrets=[SecretSource(project="myproject", path="/app")],
+                ),
+            },
+        )
+
+        with (
+            patch("psi.setup._fetch_and_register_infisical", side_effect=mock_fetch),
+            patch("psi.setup.time.sleep"),
+            pytest.raises(ProviderError, match="authentication failed"),
+        ):
+            _setup_infisical_workload(settings, "myapp", {})
+
+        assert call_count == len(_RETRY_DELAYS) + 1
+
+    def test_auth_401_wrapped_as_provider_error_not_retried(self, tmp_path: Path) -> None:
+        """Auth 401 wrapped as ProviderError is non-retryable — fails immediately."""
+        request = httpx.Request("POST", "http://test/api/v1/auth/universal-auth/login")
+        response = httpx.Response(401, request=request)
+        call_count = 0
+
+        def mock_fetch(settings, workload_name, cache_updates):
+            nonlocal call_count
+            call_count += 1
+            http_err = httpx.HTTPStatusError("401", request=request, response=response)
+            raise ProviderError(
+                "Infisical authentication failed (HTTP 401): invalid credentials",
+                provider_name="infisical",
+            ) from http_err
+
+        settings = _make_settings(
+            tmp_path,
+            workloads={
+                "myapp": WorkloadConfig(
+                    provider="infisical",
+                    secrets=[SecretSource(project="myproject", path="/app")],
+                ),
+            },
+        )
+
+        with (
+            patch("psi.setup._fetch_and_register_infisical", side_effect=mock_fetch),
+            pytest.raises(ProviderError, match="invalid credentials"),
+        ):
+            _setup_infisical_workload(settings, "myapp", {})
+
+        assert call_count == 1