fix: extension metadata from non-github sources (#4109)

lnbits/core/models/extensions.py

@@ -124,7 +124,7 @@ class ExtensionConfig(BaseModel):
     @classmethod
     async def fetch_release_config(cls, url: str) -> ExtensionConfig:
         error_msg = "Cannot fetch extension release config"
-        config = await github_api_get(url, error_msg)
+        config = await extension_metadata_get(url, error_msg)
         return ExtensionConfig.parse_obj(config)
 
     @classmethod
@@ -134,7 +134,9 @@ class ExtensionConfig(BaseModel):
         config_url = (
             f"https://raw.githubusercontent.com/{org}/{repo}/{tag_name}/config.json"
         )
-        return await cls.fetch_release_config(config_url)
+        error_msg = "Cannot fetch extension release config"
+        config = await github_api_get(config_url, error_msg)
+        return ExtensionConfig.parse_obj(config)
 
 
 class ReleasePaymentInfo(BaseModel):
@@ -1116,7 +1118,7 @@ class InstallableExtension(BaseModel):
     @classmethod
     async def fetch_manifest(cls, url) -> Manifest:
         error_msg = "Cannot fetch extensions manifest"
-        manifest = await github_api_get(url, error_msg)
+        manifest = await extension_metadata_get(url, error_msg)
         return Manifest.parse_obj(manifest)
 
 
@@ -1162,11 +1164,36 @@ class ExtensionReview(BaseModel):
     comment: str | None = Field(default=None)
 
 
+async def extension_metadata_get(url: str, error_msg: str | None) -> Any:
+    try:
+        parsed_url = httpx.URL(url)
+    except Exception as exc:
+        raise ValueError("Invalid extension metadata URL") from exc
+    if parsed_url.userinfo:
+        raise ValueError("Extension metadata URLs must not contain credentials")
+    if _is_github_token_url(url):
+        return await github_api_get(url, error_msg)
+    return await unauthenticated_json_get(url, error_msg)
+
+
+async def unauthenticated_json_get(url: str, error_msg: str | None) -> Any:
+    headers = {"User-Agent": settings.user_agent}
+    async with httpx.AsyncClient(headers=headers, follow_redirects=False) as client:
+        resp = await client.get(url)
+        if resp.status_code != 200:
+            logger.warning(f"{error_msg} ({url}): {resp.text}")
+        resp.raise_for_status()
+        return resp.json()
+
+
 async def github_api_get(url: str, error_msg: str | None) -> Any:
+    if not _is_github_token_url(url):
+        raise ValueError("Refusing GitHub authentication for an untrusted origin")
+
     headers = {"User-Agent": settings.user_agent}
     if settings.lnbits_ext_github_token:
         headers["Authorization"] = f"Bearer {settings.lnbits_ext_github_token}"
-    async with httpx.AsyncClient(headers=headers) as client:
+    async with httpx.AsyncClient(headers=headers, follow_redirects=False) as client:
         resp = await client.get(url)
         if resp.status_code != 200:
             logger.warning(f"{error_msg} ({url}): {resp.text}")
@@ -1226,3 +1253,19 @@ def _extension_manifest_sources(
     for url, manifest_type in sources:
         unique_sources.setdefault(url, manifest_type)
     return list(unique_sources.items())
+
+
+_GITHUB_TOKEN_HOSTS = frozenset({"api.github.com", "raw.githubusercontent.com"})
+
+
+def _is_github_token_url(url: str) -> bool:
+    try:
+        parsed_url = httpx.URL(url)
+    except Exception:
+        return False
+    return (
+        parsed_url.scheme == "https"
+        and parsed_url.host in _GITHUB_TOKEN_HOSTS
+        and parsed_url.port is None
+        and not parsed_url.userinfo
+    )

tests/unit/test_models_extensions.py

@@ -1,14 +1,44 @@
+import httpx
 import pytest
 from pytest_mock.plugin import MockerFixture
 
 from lnbits.core.models.extensions import (
+    ExtensionConfig,
     ExtensionManifestType,
     InstallableExtension,
     Manifest,
+    github_api_get,
 )
 from lnbits.settings import Settings
 
 
+def _mock_json_response(mocker: MockerFixture, url: str, payload: dict):
+    response = httpx.Response(
+        200,
+        json=payload,
+        request=httpx.Request("GET", url),
+    )
+    client = mocker.AsyncMock()
+    client.get.return_value = response
+    client_context = mocker.MagicMock()
+    client_context.__aenter__ = mocker.AsyncMock(return_value=client)
+    client_context.__aexit__ = mocker.AsyncMock(return_value=None)
+    client_factory = mocker.patch(
+        "lnbits.core.models.extensions.httpx.AsyncClient",
+        return_value=client_context,
+    )
+    return client_factory, client
+
+
+def _extension_config_payload() -> dict:
+    return {
+        "name": "Test Extension",
+        "short_description": "Test extension metadata",
+        "min_lnbits_version": None,
+        "max_lnbits_version": None,
+    }
+
+
 @pytest.mark.anyio
 async def test_get_installable_extensions_loads_wasm_manifests(
     settings: Settings, mocker: MockerFixture
@@ -53,3 +83,105 @@ async def test_get_installable_extensions_loads_wasm_manifests(
         regular_manifest_url,
         wasm_manifest_url,
     ]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://api.github.com/repos/example/extension",
+        "https://raw.githubusercontent.com/example/extension/main/config.json",
+    ],
+)
+async def test_release_config_sends_token_only_to_trusted_github_origins(
+    settings: Settings,
+    mocker: MockerFixture,
+    url: str,
+):
+    settings.lnbits_ext_github_token = "github-secret"
+    client_factory, client = _mock_json_response(
+        mocker, url, _extension_config_payload()
+    )
+
+    await ExtensionConfig.fetch_release_config(url)
+
+    assert client_factory.call_args.kwargs["headers"]["Authorization"] == (
+        "Bearer github-secret"
+    )
+    assert client_factory.call_args.kwargs["follow_redirects"] is False
+    client.get.assert_awaited_once_with(url)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://extensions.example/config.json",
+        "https://api.github.com.evil.example/config.json",
+        "https://api.github.com./config.json",
+        "https://raw.githubusercontent.com.evil.example/config.json",
+        "http://api.github.com/config.json",
+        "https://api.github.com:444/config.json",
+    ],
+)
+async def test_release_config_does_not_send_token_to_untrusted_origins(
+    settings: Settings,
+    mocker: MockerFixture,
+    url: str,
+):
+    settings.lnbits_ext_github_token = "github-secret"
+    client_factory, client = _mock_json_response(
+        mocker, url, _extension_config_payload()
+    )
+
+    await ExtensionConfig.fetch_release_config(url)
+
+    assert "Authorization" not in client_factory.call_args.kwargs["headers"]
+    assert client_factory.call_args.kwargs["follow_redirects"] is False
+    client.get.assert_awaited_once_with(url)
+
+
+@pytest.mark.anyio
+async def test_manifest_does_not_send_token_to_untrusted_origin(
+    settings: Settings,
+    mocker: MockerFixture,
+):
+    url = "https://extensions.example/manifest.json"
+    settings.lnbits_ext_github_token = "github-secret"
+    client_factory, client = _mock_json_response(mocker, url, {})
+
+    await InstallableExtension.fetch_manifest(url)
+
+    assert "Authorization" not in client_factory.call_args.kwargs["headers"]
+    assert client_factory.call_args.kwargs["follow_redirects"] is False
+    client.get.assert_awaited_once_with(url)
+
+
+@pytest.mark.anyio
+async def test_release_config_rejects_url_credentials(
+    settings: Settings,
+    mocker: MockerFixture,
+):
+    settings.lnbits_ext_github_token = "github-secret"
+    client_factory = mocker.patch("lnbits.core.models.extensions.httpx.AsyncClient")
+
+    with pytest.raises(ValueError, match="must not contain credentials"):
+        await ExtensionConfig.fetch_release_config(
+            "https://github-secret@api.github.com/config.json"
+        )
+
+    client_factory.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_github_api_get_rejects_untrusted_origin(
+    settings: Settings,
+    mocker: MockerFixture,
+):
+    settings.lnbits_ext_github_token = "github-secret"
+    client_factory = mocker.patch("lnbits.core.models.extensions.httpx.AsyncClient")
+
+    with pytest.raises(ValueError, match="untrusted origin"):
+        await github_api_get("https://api.github.com.evil.example/", "Cannot fetch")
+
+    client_factory.assert_not_called()