Merge pull request 'fix: batch fix all issues' (#11) from fix/all into main

Reviewed-on: https://git.home.ms/jarianc/app-store/pulls/11

android/app/src/main/kotlin/com/localstore/data/SettingsManager.kt

@@ -31,7 +31,7 @@ class SettingsManager(private val context: Context) {
         context.dataStore.data
             .catch { throw it }
             .map { prefs ->
-                prefs[PreferencesKeys.SERVER_PORT] ?: "9800"
+                prefs[PreferencesKeys.SERVER_PORT] ?: "8080"
             }
 
     fun getServerUrlFlow(): Flow<String> =

server/main.py

@@ -1,10 +1,13 @@
 import os
 import re
 import json
+import time
 import yaml
 import zipfile
 import hashlib
 import shutil
+import threading
+import asyncio
 from pathlib import Path
 from typing import Optional
 
@@ -14,7 +17,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
 
 from app import AppMetadata, AppListResponse, UpdateCheckRequest, UpdateCheckResponse
-from scanner import scan_repository, parse_manifest
+from scanner import scan_repository, parse_manifest, parse_manifest_bytes
 
 # Load config
 config_path = Path(__file__).parent / "config.yaml"
@@ -55,25 +58,29 @@ downloads_dir = Path(__file__).parent / "downloads"
 downloads_dir.mkdir(exist_ok=True)
 app.mount("/dl", StaticFiles(directory=str(downloads_dir)), name="downloads")
 
-# Cache of scanned apps
+# Cache of scanned apps (thread-safe with TTL)
 _apps_cache: list = []
-_cache_valid = False
+_cache_timestamp: float = 0.0
+_CACHE_TTL = 300  # 5 minutes
+_cache_lock = threading.Lock()
 
 
 def get_apps() -> list:
-    """Get list of apps, scanning if cache is invalid."""
+    """Get list of apps, scanning if cache is stale or invalid."""
-    global _apps_cache, _cache_valid
+    global _apps_cache, _cache_timestamp
-    if not _cache_valid:
+    with _cache_lock:
+        if not _apps_cache or (time.time() - _cache_timestamp) > _CACHE_TTL:
             _apps_cache = scan_repository(str(REPO_PATH))
-        _cache_valid = True
+            _cache_timestamp = time.time()
-        print(f"Repository scanned: {_apps_cache.__len__()} apps found")
+            print(f"Repository scanned: {len(_apps_cache)} apps found")
     return _apps_cache
 
 
 def invalidate_cache():
     """Invalidate the apps cache."""
-    global _cache_valid
+    global _cache_timestamp
-    _cache_valid = False
+    with _cache_lock:
+        _cache_timestamp = 0
 
 
 def find_app(app_id: str) -> Optional[dict]:
@@ -227,8 +234,12 @@ def get_app_screenshots(app_id: str):
 @app.get("/api/apps/{app_id}/screenshots/{filename}")
 def get_screenshot(app_id: str, filename: str):
     """Get a specific screenshot."""
+    safe_filename = os.path.basename(filename)
     screenshots_dir = REPO_PATH / f"{app_id}_screenshots"
-    screenshot_path = screenshots_dir / filename
+    screenshot_path = (screenshots_dir / safe_filename).resolve()
+
+    if not str(screenshot_path).startswith(str(screenshots_dir.resolve())):
+        raise HTTPException(status_code=400, detail="Invalid path")
 
     if not screenshot_path.exists():
         raise HTTPException(status_code=404, detail="Screenshot not found")
@@ -285,18 +296,22 @@ async def upload_app(
     if not apk.filename or not apk.filename.endswith(".apk"):
         raise HTTPException(400, "File must be an APK")
 
+    contents = await apk.read()
+    if len(contents) < 4 or contents[:4] != b"PK\x03\x04":
+        raise HTTPException(400, "File is not a valid APK (invalid ZIP magic bytes)")
+
     target_id = app_id or _sanitize_app_id(apk.filename)
     apk_target = REPO_PATH / f"{target_id}.apk"
 
     # Check for duplicate
     if apk_target.exists():
         existing = parse_manifest(str(apk_target))
-        new = parse_manifest(str(apk.filename))
+        new = parse_manifest_bytes(contents)
         if existing.get("version_code") == new.get("version_code"):
             raise HTTPException(409, f"App '{target_id}' already exists with same version")
 
     # Save APK
-    apk_target.write_bytes(await apk.read())
+    apk_target.write_bytes(contents)
 
     # Save metadata JSON if provided
     if metadata and metadata.filename and metadata.filename.endswith(".json"):

server/scanner.py

@@ -1,5 +1,6 @@
 import os
 import re
+import subprocess
 import zipfile
 import xml.etree.ElementTree as ET
 from typing import Dict, List, Optional
@@ -23,9 +24,8 @@ def parse_manifest(apk_path: str) -> Dict:
         with zipfile.ZipFile(apk_path) as z:
             # Check for compiled manifest (binary XML)
             if "AndroidManifest.xml" in z.namelist():
-                # Binary XML - try to extract what we can
+                # Binary XML - use aapt for reliable parsing
-                data = z.read("AndroidManifest.xml")
+                metadata = _parse_binary_manifest(apk_path, metadata)
-                metadata = _parse_binary_manifest(data, metadata)
             else:
                 # No manifest found
                 return metadata
@@ -35,7 +35,8 @@ def parse_manifest(apk_path: str) -> Dict:
                 apk_name = Path(apk_path).stem
                 metadata["package_name"] = apk_name
 
-            # Look for icon resource
+            # Look for icon resource (fallback if aapt didn't find one)
+            if not metadata.get("icon"):
                 icon_res = _find_icon_resource(apk_path)
                 if icon_res:
                     metadata["icon"] = icon_res
@@ -46,126 +47,174 @@ def parse_manifest(apk_path: str) -> Dict:
     return metadata
 
 
-def _parse_binary_manifest(data: bytes, metadata: Dict) -> Dict:
+def parse_manifest_bytes(data: bytes) -> Dict:
-    """Parse binary XML manifest. Android uses a proprietary binary XML format."""
+    """Extract metadata from APK bytes (for in-memory validation)."""
-    try:
+    metadata = {
-        # Try to decode string table - the binary XML format has a header
+        "package_name": None,
-        # then a string table, then keypool, then resources
+        "version_name": None,
-        # This is a simplified parser that extracts common fields
+        "version_code": None,
+        "label": None,
+        "icon": None,
+        "min_sdk": None,
+        "target_sdk": None,
+        "permissions": [],
+    }
+
+    try:
+        import tempfile
+        with tempfile.NamedTemporaryFile(suffix=".apk", delete=False) as tmp:
+            tmp.write(data)
+            tmp_path = tmp.name
+        try:
+            with zipfile.ZipFile(tmp_path) as z:
+                if "AndroidManifest.xml" in z.namelist():
+                    metadata = _parse_binary_manifest(tmp_path, metadata)
+        finally:
+            os.unlink(tmp_path)
+    except Exception as e:
+        print(f"Error parsing APK bytes: {e}")
 
-        # Check if it's binary XML (starts with XML header magic)
-        if len(data) < 20:
     return metadata
 
-        # Binary XML header magic: 0x00080003
-        if data[0:4] == b'\x03\x00\x08\x00' or data[0:4] == b'\x00\x08\x00\x03':
-            # Try to extract strings from the string table
-            strings = _extract_strings(data)
 
-            # Look for package attribute
+def _parse_binary_manifest(data_or_path, metadata: Dict) -> Dict:
+    """Parse binary XML manifest using aapt for reliable extraction.
+
+    Accepts either a file path (str) or raw APK bytes.
+    """
+    import tempfile
+
+    if isinstance(data_or_path, str):
+        # Already a file path, use directly with aapt
+        result = _parse_with_aapt(data_or_path)
+        if result:
+            metadata.update(result)
+        return metadata
+
+    # Raw bytes - write temp APK for aapt
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".apk", delete=False) as tmp:
+            tmp.write(data_or_path)
+            tmp_path = tmp.name
+
+        result = _parse_with_aapt(tmp_path)
+        os.unlink(tmp_path)
+
+        if result:
+            metadata.update(result)
+    except Exception as e:
+        print(f"Error parsing binary manifest with aapt: {e}")
+        # Fallback: basic string extraction if aapt fails
+        _parse_fallback(data_or_path, metadata)
+
+    return metadata
+
+
+def _parse_with_aapt(apk_path: str) -> Optional[Dict]:
+    """Use aapt or aapt2 to extract manifest info reliably."""
+    for tool in ["aapt2", "aapt"]:
+        try:
+            result = subprocess.run(
+                [tool, "dump", "badging", apk_path],
+                capture_output=True, text=True, timeout=30
+            )
+            if result.returncode == 0:
+                return _parse_aapt_output(result.stdout)
+        except (FileNotFoundError, subprocess.TimeoutExpired):
+            continue
+    return None
+
+
+def _parse_aapt_output(output: str) -> Dict:
+    """Parse aapt dump badging output."""
+    result = {
+        "package_name": None,
+        "version_name": None,
+        "version_code": None,
+        "label": None,
+        "icon": None,
+        "min_sdk": None,
+        "target_sdk": None,
+        "permissions": [],
+    }
+
+    for line in output.splitlines():
+        # Package: name='com.example.app'
+        m = re.search(r"package: name='([^']+)'", line)
+        if m:
+            result["package_name"] = m.group(1)
+        # versionName='1.0.0'
+        m = re.search(r"versionName='([^']+)'", line)
+        if m:
+            result["version_name"] = m.group(1)
+        # versionCode='1'
+        m = re.search(r"versionCode='([^']+)'", line)
+        if m:
+            result["version_code"] = m.group(1)
+        # application-label:'My App'
+        m = re.search(r"application-label:'([^']*)'", line)
+        if m:
+            result["label"] = m.group(1)
+        # application-icon-128:'res/path'
+        m = re.search(r"application-icon-\d+:'([^']+)'", line)
+        if m:
+            result["icon"] = m.group(1)
+        # sdkVersion:'21' / minSdkVersion:'21' / targetSdkVersion:'33'
+        m = re.search(r"minSdkVersion='([^']+)'", line)
+        if m:
+            result["min_sdk"] = m.group(1)
+        m = re.search(r"targetSdkVersion='([^']+)'", line)
+        if m:
+            result["target_sdk"] = m.group(1)
+        # uses-permission: name='android.permission.INTERNET'
+        m = re.search(r"uses-permission: name='([^']+)'", line)
+        if m:
+            result["permissions"].append(m.group(1))
+
+    return result
+
+
+def _parse_fallback(data: bytes, metadata: Dict) -> None:
+    """Fallback string-based extraction when aapt is unavailable."""
+    try:
+        if len(data) < 20:
+            return
+
+        if data[0:4] != b'\x03\x00\x08\x00' and data[0:4] != b'\x00\x08\x00\x03':
+            return
+
+        strings = _extract_strings_fallback(data)
+
         for s in strings:
             if s and "." in s and not s.startswith("android"):
-                    # Likely a package name
                 if re.match(r'^[a-zA-Z][a-zA-Z0-9_.]*$', s) and len(s.split('.')) >= 2:
+                    if metadata["package_name"] is None:
                         metadata["package_name"] = s
                         break
 
-            # Look for versionName
         for s in strings:
             if s and re.match(r'^\d+\.\d+', s):
+                if metadata["version_name"] is None:
                     metadata["version_name"] = s
                     break
 
-            # Look for versionCode (numeric)
-            for s in strings:
-                if s and re.match(r'^\d{1,10}$', s) and len(s) < 10:
-                    # Could be version code or other number
-                    if metadata["version_code"] is None:
-                        metadata["version_code"] = s
-                        break
-
-            # Look for app label - usually contains spaces or title-case words
-            for s in strings:
-                if s and len(s) > 1 and len(s) < 50 and not s.startswith("android"):
-                    if any(c.isupper() for c in s) or " " in s:
-                        if s != metadata["package_name"]:
-                            metadata["label"] = s
-                            break
-
     except Exception as e:
-        print(f"Error parsing binary manifest: {e}")
+        print(f"Error in fallback manifest parsing: {e}")
-
-    return metadata
 
 
-def _extract_strings(data: bytes) -> List[str]:
+def _extract_strings_fallback(data: bytes) -> List[str]:
-    """Extract string table from binary XML."""
+    """Extract printable strings >= 4 chars from binary data."""
     strings = []
-    try:
+    current = []
-        # Binary XML header structure:
+    for byte in data:
-        # 4 bytes: type (0x0008)
+        if 32 <= byte < 127:
-        # 4 bytes: reserved
+            current.append(chr(byte))
-        # 4 bytes: header size
+        else:
-        # 4 bytes: end of strings offset
+            if len(current) >= 4:
-        # 4 bytes: start of string data offset
+                strings.append(''.join(current))
-
+            current = []
-        if len(data) < 20:
+    if len(current) >= 4:
-            return strings
+        strings.append(''.join(current))
-
-        header_size = int.from_bytes(data[8:12], 'little')
-        strings_end = int.from_bytes(data[12:16], 'little')
-        strings_start = int.from_bytes(data[16:20], 'little')
-
-        if strings_end <= strings_start or strings_start < header_size:
-            return strings
-
-        # String count is at offset header_size (right after header)
-        str_count_offset = header_size
-        if str_count_offset + 4 > len(data):
-            return strings
-
-        str_count = int.from_bytes(data[str_count_offset:str_count_offset + 4], 'little')
-
-        # Index table starts after count
-        index_offset = str_count_offset + 4
-
-        # Extract string indices
-        indices = []
-        for i in range(min(str_count, 1000)):  # limit to avoid issues
-            idx_off = index_offset + (i * 4)
-            if idx_off + 4 > len(data):
-                break
-            idx = int.from_bytes(data[idx_off:idx_off + 4], 'little')
-            indices.append(idx)
-
-        # Extract strings using indices
-        for idx in indices:
-            str_offset = strings_start + idx
-            if str_offset + 4 > len(data):
-                continue
-
-            # String length is stored as 2 bytes (UTF-16 length)
-            utf16_len = int.from_bytes(data[str_offset:str_offset + 2], 'little')
-            # Then 2 bytes for UTF-8 length
-            utf8_len = int.from_bytes(data[str_offset + 2:str_offset + 4], 'little')
-
-            if utf8_len == 0 or utf8_len > 500:
-                continue
-
-            str_data_start = str_offset + 4
-            if str_data_start + utf8_len > len(data):
-                continue
-
-            try:
-                s = data[str_data_start:str_data_start + utf8_len].decode('utf-8', errors='ignore')
-                if s:
-                    strings.append(s)
-            except Exception:
-                continue
-
-    except Exception as e:
-        print(f"Error extracting strings: {e}")
-
     return strings