fix pyright lnbits

Co-authored-by: dni ⚡ <office@dnilabs.com>
2023-04-04 07:34:17 +02:00
parent 3855cf47f3
commit 02306148df
8 changed files with 266 additions and 116 deletions
@@ -3,20 +3,146 @@ import json
 import os
 import shutil
 import sys
-import urllib.request
 import zipfile
 from http import HTTPStatus
 from pathlib import Path
 from typing import Any, List, NamedTuple, Optional, Tuple
+from urllib import request

 import httpx
-from fastapi.exceptions import HTTPException
+from fastapi import HTTPException
+from fastapi.responses import JSONResponse
 from loguru import logger
 from pydantic import BaseModel

 from lnbits.settings import settings


+class ExplicitRelease(BaseModel):
+    id: str
+    name: str
+    version: str
+    archive: str
+    hash: str
+    dependencies: List[str] = []
+    icon: Optional[str]
+    short_description: Optional[str]
+    html_url: Optional[str]
+    details: Optional[str]
+    info_notification: Optional[str]
+    critical_notification: Optional[str]
+
+
+class GitHubRelease(BaseModel):
+    id: str
+    organisation: str
+    repository: str
+
+
+class Manifest(BaseModel):
+    featured: List[str] = []
+    extensions: List["ExplicitRelease"] = []
+    repos: List["GitHubRelease"] = []
+
+
+class GitHubRepoRelease(BaseModel):
+    name: str
+    tag_name: str
+    zipball_url: str
+    html_url: str
+
+
+class GitHubRepo(BaseModel):
+    stargazers_count: str
+    html_url: str
+    default_branch: str
+
+
+class ExtensionConfig(BaseModel):
+    name: str
+    short_description: str
+    tile: str = ""
+
+
+def download_url(url, save_path):
+    with request.urlopen(url) as dl_file:
+        with open(save_path, "wb") as out_file:
+            out_file.write(dl_file.read())
+
+
+def file_hash(filename):
+    h = hashlib.sha256()
+    b = bytearray(128 * 1024)
+    mv = memoryview(b)
+    with open(filename, "rb", buffering=0) as f:
+        while n := f.readinto(mv):
+            h.update(mv[:n])
+    return h.hexdigest()
+
+
+async def fetch_github_repo_info(
+    org: str, repository: str
+) -> Tuple[GitHubRepo, GitHubRepoRelease, ExtensionConfig]:
+    repo_url = f"https://api.github.com/repos/{org}/{repository}"
+    error_msg = "Cannot fetch extension repo"
+    repo = await gihub_api_get(repo_url, error_msg)
+    github_repo = GitHubRepo.parse_obj(repo)
+
+    lates_release_url = (
+        f"https://api.github.com/repos/{org}/{repository}/releases/latest"
+    )
+    error_msg = "Cannot fetch extension releases"
+    latest_release: Any = await gihub_api_get(lates_release_url, error_msg)
+
+    config_url = f"https://raw.githubusercontent.com/{org}/{repository}/{github_repo.default_branch}/config.json"
+    error_msg = "Cannot fetch config for extension"
+    config = await gihub_api_get(config_url, error_msg)
+
+    return (
+        github_repo,
+        GitHubRepoRelease.parse_obj(latest_release),
+        ExtensionConfig.parse_obj(config),
+    )
+
+
+async def fetch_manifest(url) -> Manifest:
+    error_msg = "Cannot fetch extensions manifest"
+    manifest = await gihub_api_get(url, error_msg)
+    return Manifest.parse_obj(manifest)
+
+
+async def fetch_github_releases(org: str, repo: str) -> List[GitHubRepoRelease]:
+    releases_url = f"https://api.github.com/repos/{org}/{repo}/releases"
+    error_msg = "Cannot fetch extension releases"
+    releases = await gihub_api_get(releases_url, error_msg)
+    return [GitHubRepoRelease.parse_obj(r) for r in releases]
+
+
+async def gihub_api_get(url: str, error_msg: Optional[str]) -> Any:
+    async with httpx.AsyncClient() as client:
+        headers = (
+            {"Authorization": "Bearer " + settings.lnbits_ext_github_token}
+            if settings.lnbits_ext_github_token
+            else None
+        )
+        resp = await client.get(
+            url,
+            headers=headers,
+        )
+        if resp.status_code != 200:
+            logger.warning(f"{error_msg} ({url}): {resp.text}")
+        resp.raise_for_status()
+        return resp.json()
+
+
+def icon_to_github_url(source_repo: str, path: Optional[str]) -> str:
+    if not path:
+        return ""
+    _, _, *rest = path.split("/")
+    tail = "/".join(rest)
+    return f"https://github.com/{source_repo}/raw/main/{tail}"
+
+
 class Extension(NamedTuple):
    code: str
    is_valid: bool
@@ -97,12 +223,12 @@ class ExtensionRelease(BaseModel):
    version: str
    archive: str
    source_repo: str
-    is_github_release = False
-    hash: Optional[str]
-    html_url: Optional[str]
-    description: Optional[str]
+    is_github_release: bool = False
+    hash: Optional[str] = None
+    html_url: Optional[str] = None
+    description: Optional[str] = None
    details_html: Optional[str] = None
-    icon: Optional[str]
+    icon: Optional[str] = None

    @classmethod
    def from_github_release(
@@ -132,52 +258,6 @@ class ExtensionRelease(BaseModel):
            return []


-class ExplicitRelease(BaseModel):
-    id: str
-    name: str
-    version: str
-    archive: str
-    hash: str
-    dependencies: List[str] = []
-    icon: Optional[str]
-    short_description: Optional[str]
-    html_url: Optional[str]
-    details: Optional[str]
-    info_notification: Optional[str]
-    critical_notification: Optional[str]
-
-
-class GitHubRelease(BaseModel):
-    id: str
-    organisation: str
-    repository: str
-
-
-class Manifest(BaseModel):
-    featured: List[str] = []
-    extensions: List["ExplicitRelease"] = []
-    repos: List["GitHubRelease"] = []
-
-
-class GitHubRepoRelease(BaseModel):
-    name: str
-    tag_name: str
-    zipball_url: str
-    html_url: str
-
-
-class GitHubRepo(BaseModel):
-    stargazers_count: str
-    html_url: str
-    default_branch: str
-
-
-class ExtensionConfig(BaseModel):
-    name: str
-    short_description: str
-    tile: str = ""
-
-
 class InstallableExtension(BaseModel):
    id: str
    name: str
@@ -187,8 +267,9 @@ class InstallableExtension(BaseModel):
    is_admin_only: bool = False
    stars: int = 0
    featured = False
-    latest_release: Optional[ExtensionRelease]
-    installed_release: Optional[ExtensionRelease]
+    latest_release: Optional[ExtensionRelease] = None
+    installed_release: Optional[ExtensionRelease] = None
+    archive: Optional[str] = None

    @property
    def hash(self) -> str:
@@ -234,6 +315,7 @@ class InstallableExtension(BaseModel):
        if ext_zip_file.is_file():
            os.remove(ext_zip_file)
        try:
+            assert self.installed_release
            download_url(self.installed_release.archive, ext_zip_file)
        except Exception as ex:
            logger.warning(ex)
@@ -334,8 +416,7 @@ class InstallableExtension(BaseModel):
                id=github_release.id,
                name=config.name,
                short_description=config.short_description,
-                version="0",
-                stars=repo.stargazers_count,
+                stars=int(repo.stargazers_count),
                icon=icon_to_github_url(
                    f"{github_release.organisation}/{github_release.repository}",
                    config.tile,
@@ -354,7 +435,6 @@ class InstallableExtension(BaseModel):
            id=e.id,
            name=e.name,
            archive=e.archive,
-            hash=e.hash,
            short_description=e.short_description,
            icon=e.icon,
            dependencies=e.dependencies,
@@ -443,6 +523,52 @@ class InstallableExtension(BaseModel):
        return selected_release[0] if len(selected_release) != 0 else None


+class InstalledExtensionMiddleware:
+    # This middleware class intercepts calls made to the extensions API and:
+    #  - it blocks the calls if the extension has been disabled or uninstalled.
+    #  - it redirects the calls to the latest version of the extension if the extension has been upgraded.
+    #  - otherwise it has no effect
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if "path" not in scope:
+            await self.app(scope, receive, send)
+            return
+
+        path_elements = scope["path"].split("/")
+        if len(path_elements) > 2:
+            _, path_name, path_type, *rest = path_elements
+            tail = "/".join(rest)
+        else:
+            _, path_name = path_elements
+            path_type = None
+            tail = ""
+
+        # block path for all users if the extension is disabled
+        if path_name in settings.lnbits_deactivated_extensions:
+            response = JSONResponse(
+                status_code=HTTPStatus.NOT_FOUND,
+                content={"detail": f"Extension '{path_name}' disabled"},
+            )
+            await response(scope, receive, send)
+            return
+
+        # re-route API trafic if the extension has been upgraded
+        if path_type == "api":
+            upgraded_extensions = list(
+                filter(
+                    lambda ext: ext.endswith(f"/{path_name}"),
+                    settings.lnbits_upgraded_extensions,
+                )
+            )
+            if len(upgraded_extensions) != 0:
+                upgrade_path = upgraded_extensions[0]
+                scope["path"] = f"/upgrades/{upgrade_path}/{path_type}/{tail}"
+
+        await self.app(scope, receive, send)
+
+
 class CreateExtension(BaseModel):
    ext_id: str
    archive: str