feat(storage): implement support for storing artifacts to S3 (closes

Author: jkuri

README.md

@@ -25,6 +25,7 @@ A simplistic, opinionated remote update server implementing hawkBit™'s [DDI AP
     ```
 
 3. Launch gooseBit:
+
     ```bash
     python main.py
     ```
@@ -140,16 +141,17 @@ The structure of gooseBit is as follows:
 
 - `api`: Files for the API.
 - `ui`: Files for the UI.
-    - `bff`: Backend for frontend API.
-    - `static`: Static files.
-    - `templates`: Jinja2 formatted templates.
-    - `nav`: Navbar handler.
+  - `bff`: Backend for frontend API.
+  - `static`: Static files.
+  - `templates`: Jinja2 formatted templates.
+  - `nav`: Navbar handler.
 - `updater`: DDI API handler and device update manager.
 - `updates`: SWUpdate file parsing.
 - `auth`: Authentication functions and permission handling.
 - `models`: Database models.
 - `db`: Database config and initialization.
 - `schema`: Pydantic models used for API type hinting.
 - `settings`: Settings loader and handler.
+- `storage`: Storage for software artifacts.
 - `telemetry`: Telemetry data handlers.
 - `routes`: Routes for a giving endpoint, including the router.

conftest.py

@@ -36,14 +36,22 @@ async def clear_cache():
 @pytest_asyncio.fixture(scope="function")
 async def test_app():
     from goosebit.users import create_initial_user
+    from goosebit.storage import init_storage, close_storage
+    from goosebit.settings import config
 
     async with RegisterTortoise(
         app=app,
         config=TORTOISE_CONF,
     ):
         await Tortoise.generate_schemas()
         await create_initial_user(username="testing@goosebit.test", hashed_pwd=PWD_CXT.hash("test"))
-        yield app
+
+        init_storage(config)
+
+        try:
+            yield app
+        finally:
+            close_storage()
 
 
 @pytest_asyncio.fixture(scope="function")

goosebit.yaml

@@ -68,3 +68,13 @@ logging:
   root:
     level: INFO
     handlers: [console]
+
+# Storage settings (default backend is filesystem)
+# storage:
+#   backend: s3
+#   s3:
+#     bucket: goosebit
+#     # region: us-east-1
+#     # endpoint_url: http://localhost:9000 # example for self-hosted min.io
+#     access_key_id: minioadmin
+#     secret_access_key: minioadmin

goosebit/__init__.py

@@ -16,6 +16,7 @@
 from goosebit import api, db, ui, updater
 from goosebit.auth import get_user_from_request, login_user, redirect_if_authenticated
 from goosebit.settings import PWD_CXT, config
+from goosebit.storage import init_storage, close_storage
 from goosebit.ui.nav import nav
 from goosebit.ui.static import static
 from goosebit.ui.templates import templates
@@ -29,9 +30,15 @@ async def lifespan(_: FastAPI):
     db_ready = await db.init()
     if not db_ready:
         logger.exception("DB does not exist, try running `poetry run aerich upgrade`.")
+
+    init_storage(config)
+    logger.debug(f"Initialized storage backend: {config.storage.backend}")
+
     if db_ready:
         yield
+
     await db.close()
+    close_storage()
 
 
 app = FastAPI(

goosebit/api/v1/download/routes.py

@@ -1,8 +1,9 @@
 from fastapi import APIRouter, HTTPException
 from fastapi.requests import Request
-from fastapi.responses import FileResponse, RedirectResponse
+from fastapi.responses import FileResponse, RedirectResponse, StreamingResponse
 
 from goosebit.db.models import Software
+from goosebit.storage import get_storage
 
 router = APIRouter(prefix="/download", tags=["download"])
 
@@ -18,5 +19,16 @@ async def download_file(_: Request, file_id: int):
             media_type="application/octet-stream",
             filename=software.path.name,
         )
-    else:
-        return RedirectResponse(url=software.uri)
+
+    storage = get_storage()
+    try:
+        url = await storage.get_download_url(software.uri)
+        return RedirectResponse(url=url)
+    except Exception:
+        # Fallback to streaming if redirect fails.
+        file_stream = storage.get_file_stream(software.uri)
+        return StreamingResponse(
+            file_stream,
+            media_type="application/octet-stream",
+            headers={"Content-Disposition": f"attachment; filename={software.path.name}"}
+        )

goosebit/settings/schema.py

@@ -34,6 +34,21 @@ class PrometheusSettings(BaseModel):
 class MetricsSettings(BaseModel):
     prometheus: PrometheusSettings = PrometheusSettings()
 
+class StorageType(StrEnum):
+    FILESYSTEM = "filesystem"
+    S3 = "s3"
+
+class S3StorageSettings(BaseModel):
+    bucket: str
+    region: str = "us-east-1"
+    endpoint_url: str | None = None
+    access_key_id: str | None = None
+    secret_access_key: str | None = None
+
+
+class StorageSettings(BaseModel):
+    backend: StorageType = StorageType.FILESYSTEM
+    s3: S3StorageSettings | None = None
 
 class GooseBitSettings(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="GOOSEBIT_")
@@ -54,6 +69,8 @@ class GooseBitSettings(BaseSettings):
 
     artifacts_dir: Path = GOOSEBIT_ROOT_DIR.joinpath("artifacts")
 
+    storage: StorageSettings = StorageSettings()
+
     metrics: MetricsSettings = MetricsSettings()
 
     logging: dict = LOGGING_DEFAULT

goosebit/storage/__init__.py

@@ -0,0 +1,60 @@
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import AsyncGenerator
+
+from goosebit.settings.schema import GooseBitSettings, StorageType
+
+
+class Storage(ABC):
+    @abstractmethod
+    async def store_file(self, source_path: Path, key: str) -> str:
+        pass
+
+    @abstractmethod
+    async def get_file_stream(self, uri: str) -> AsyncGenerator[bytes, None]:
+        pass
+
+    @abstractmethod
+    async def get_download_url(self, uri: str) -> str:
+        pass
+
+
+_storage: Storage | None = None
+
+
+def init_storage(config: GooseBitSettings) -> None:
+    global _storage
+    _storage = create_storage(config)
+
+def create_storage(config: GooseBitSettings) -> Storage:
+    from .filesystem import FilesystemStorage
+    from .s3 import S3Storage
+
+    if config.storage.backend == StorageType.FILESYSTEM:
+        return FilesystemStorage(base_path=config.artifacts_dir)
+
+    elif config.storage.backend == StorageType.S3:
+        if config.storage.s3 is None:
+            return FilesystemStorage(base_path=config.artifacts_dir)
+
+        config = config.storage.s3
+        return S3Storage(
+            bucket=config.bucket,
+            region=config.region,
+            endpoint_url=config.endpoint_url,
+            access_key_id=config.access_key_id,
+            secret_access_key=config.secret_access_key
+        )
+
+    else:
+        raise ValueError(f"Unknown storage backend type: {config.storage.backend}")
+
+def get_storage() -> Storage:
+    if _storage is None:
+        raise RuntimeError("Storage backend not initialized. Call init_storage() first.")
+
+    return _storage
+
+def close_storage() -> None:
+    global _storage
+    _storage = None

goosebit/storage/filesystem.py

@@ -0,0 +1,71 @@
+import shutil
+from pathlib import Path
+from typing import AsyncGenerator
+from urllib.parse import urlparse
+
+import httpx
+from anyio import open_file
+
+from . import Storage
+
+
+class FilesystemStorage(Storage):
+    def __init__(self, base_path: Path):
+        self.base_path = Path(base_path)
+        self.base_path.mkdir(parents=True, exist_ok=True)
+
+    async def store_file(self, source_path: Path, key: str) -> str:
+        dest_path = self.base_path / key
+        dest_path.parent.mkdir(parents=True, exist_ok=True)
+
+        shutil.copy2(source_path, dest_path)
+
+        return dest_path.resolve().as_uri()
+
+    async def get_file_stream(self, uri: str) -> AsyncGenerator[bytes, None]:
+        parsed = urlparse(uri)
+
+        if parsed.scheme in ("http", "https"):
+            async with httpx.AsyncClient() as client:
+                async with client.stream("GET", uri) as response:
+                    response.raise_for_status()
+                    async for chunk in response.aiter_bytes(8192):
+                        yield chunk
+
+        elif parsed.scheme == "file":
+            file_path = self._extract_path_from_uri(uri)
+            if not file_path.exists():
+                raise FileNotFoundError(f"File not found: {file_path}")
+
+            async with await open_file(file_path, "rb") as f:
+                while True:
+                    chunk = await f.read(8192)
+                    if not chunk:
+                        break
+                    yield chunk
+        else:
+            raise ValueError(f"Unsupported URI scheme '{parsed.scheme}' for filesystem backend: {uri}")
+
+    async def get_download_url(self, uri: str) -> str:
+        parsed = urlparse(uri)
+
+        if parsed.scheme in ("http", "https"):
+            return uri
+
+        elif parsed.scheme == "file":
+            file_path = self._extract_path_from_uri(uri)
+            if not file_path.exists():
+                raise FileNotFoundError(f"File not found: {file_path}")
+
+            return file_path.resolve().as_uri()
+
+        else:
+            raise ValueError(f"Unsupported URI scheme '{parsed.scheme}' for filesystem backend: {uri}")
+
+    def _extract_path_from_uri(self, uri: str) -> Path:
+        parsed = urlparse(uri)
+
+        if parsed.scheme != "file":
+            raise ValueError(f"Expected file:// URI, got: {uri}")
+
+        return Path(parsed.path)

goosebit/storage/s3.py

@@ -0,0 +1,98 @@
+import asyncio
+from pathlib import Path
+from typing import AsyncGenerator
+
+from boto3.session import Session
+from botocore.config import Config
+from botocore.exceptions import ClientError
+
+from . import Storage
+
+class S3Storage(Storage):
+    def __init__(
+        self,
+        bucket: str,
+        region: str = "us-east-1",
+        endpoint_url: str | None = None,
+        access_key_id: str | None = None,
+        secret_access_key: str | None = None
+    ):
+        self.bucket = bucket
+
+        config = Config(
+            region_name = region,
+            connect_timeout = 10,
+            read_timeout = 60,
+            retries = {"max_attempts": 5, "mode": "adaptive"},
+            signature_version = "s3v4",
+        )
+
+        session_config = {}
+        if access_key_id is not None:
+            session_config["aws_access_key_id"] = access_key_id
+        if secret_access_key is not None:
+            session_config["aws_secret_access_key"] = secret_access_key
+
+        session = Session(**session_config)
+
+        self.s3_client = session.client("s3", config=config, endpoint_url=endpoint_url)
+
+    async def store_file(self, source_path: Path, key: str) -> str:
+        try:
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(
+                None,
+                self.s3_client.upload_file,
+                str(source_path),
+                self.bucket,
+                key
+            )
+            return f"s3://{self.bucket}/{key}"
+        except ClientError as e:
+            raise ValueError(f"S3 upload failed: {e}")
+
+    async def get_file_stream(self, uri: str) -> AsyncGenerator[bytes, None]:
+        key = self._extract_key_from_uri(uri)
+
+        try:
+            loop = asyncio.get_event_loop()
+            response = await loop.run_in_executor(
+                None,
+                self.s3_client.get_object,
+                self.bucket,
+                key
+            )
+
+            body = response["Body"]
+            try:
+                while True:
+                    chunk = await loop.run_in_executor(None, body.read, 8192)
+                    if not chunk:
+                        break
+                    yield chunk
+            finally:
+                await loop.run_in_executor(None, body.close)
+
+        except ClientError as e:
+            raise ValueError(f"S3 download failed: {e}")
+
+    async def get_download_url(self, uri: str) -> str:
+        key = self._extract_key_from_uri(uri)
+
+        try:
+            loop = asyncio.get_event_loop()
+            return await loop.run_in_executor(
+                None,
+                self.s3_client.generate_presigned_url,
+                "get_object",
+                {"Bucket": self.bucket, "Key": key},
+                3600 # 1 hour expiration
+            )
+        except ClientError as e:
+            raise ValueError(f"Failed to generate presigned URL: {e}")
+
+    def _extract_key_from_uri(self, uri: str) -> str:
+        if not uri.startswith(f"s3://{self.bucket}/"):
+            raise ValueError(f"Invalid S3 URI for bucket {self.bucket}: {uri}")
+
+        return uri.replace(f"s3://{self.bucket}/", "")