From 10fbfb5c5ae7775149ec63f4721f1fe67ddc1456 Mon Sep 17 00:00:00 2001 From: Jarian Date: Sun, 5 Jul 2026 22:55:30 +0000 Subject: [PATCH] fix: batch fix all open issues - scanner.py: replace heuristic binary manifest parser with aapt/aapt2 (fixes #2, #7) - main.py: add ZIP magic bytes validation on APK upload (fixes #8) - main.py: sanitize screenshot filename with os.path.basename + resolve() check (fixes #6) - main.py: add threading.Lock + TTL to in-memory cache (fixes #4, #9) - SettingsManager.kt: default port 9800 -> 8080 to match config.yaml (fixes #3) - main.py: parse_manifest_bytes for in-memory APK validation on upload --- .../com/localstore/data/SettingsManager.kt | 2 +- server/main.py | 43 ++- server/scanner.py | 271 +++++++++++------- 3 files changed, 190 insertions(+), 126 deletions(-) diff --git a/android/app/src/main/kotlin/com/localstore/data/SettingsManager.kt b/android/app/src/main/kotlin/com/localstore/data/SettingsManager.kt index dee09c7..ccfd71c 100644 --- a/android/app/src/main/kotlin/com/localstore/data/SettingsManager.kt +++ b/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 = diff --git a/server/main.py b/server/main.py index 2c813b6..403a446 100644 --- a/server/main.py +++ b/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.""" - global _apps_cache, _cache_valid - if not _cache_valid: - _apps_cache = scan_repository(str(REPO_PATH)) - _cache_valid = True - print(f"Repository scanned: {_apps_cache.__len__()} apps found") + """Get list of apps, scanning if cache is stale or invalid.""" + global _apps_cache, _cache_timestamp + with _cache_lock: + if not _apps_cache or (time.time() - _cache_timestamp) > _CACHE_TTL: + _apps_cache = scan_repository(str(REPO_PATH)) + _cache_timestamp = time.time() + print(f"Repository scanned: {len(_apps_cache)} apps found") return _apps_cache def invalidate_cache(): """Invalidate the apps cache.""" - global _cache_valid - _cache_valid = False + global _cache_timestamp + 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"): diff --git a/server/scanner.py b/server/scanner.py index fe4d4e5..1ced99c 100644 --- a/server/scanner.py +++ b/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 - data = z.read("AndroidManifest.xml") - metadata = _parse_binary_manifest(data, metadata) + # Binary XML - use aapt for reliable parsing + metadata = _parse_binary_manifest(apk_path, metadata) else: # No manifest found return metadata @@ -35,10 +35,11 @@ def parse_manifest(apk_path: str) -> Dict: apk_name = Path(apk_path).stem metadata["package_name"] = apk_name - # Look for icon resource - icon_res = _find_icon_resource(apk_path) - if icon_res: - metadata["icon"] = icon_res + # 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 except Exception as e: print(f"Error parsing {apk_path}: {e}") @@ -46,126 +47,174 @@ def parse_manifest(apk_path: str) -> Dict: return metadata -def _parse_binary_manifest(data: bytes, metadata: Dict) -> Dict: - """Parse binary XML manifest. Android uses a proprietary binary XML format.""" +def parse_manifest_bytes(data: bytes) -> Dict: + """Extract metadata from APK bytes (for in-memory validation).""" + metadata = { + "package_name": None, + "version_name": None, + "version_code": None, + "label": None, + "icon": None, + "min_sdk": None, + "target_sdk": None, + "permissions": [], + } + try: - # Try to decode string table - the binary XML format has a header - # then a string table, then keypool, then resources - # This is a simplified parser that extracts common fields - - # 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 - 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: - metadata["package_name"] = s - break - - # Look for versionName - for s in strings: - if s and re.match(r'^\d+\.\d+', s): - 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 - + 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 binary manifest: {e}") + print(f"Error parsing APK bytes: {e}") return metadata -def _extract_strings(data: bytes) -> List[str]: - """Extract string table from binary XML.""" - strings = [] +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: - # Binary XML header structure: - # 4 bytes: type (0x0008) - # 4 bytes: reserved - # 4 bytes: header size - # 4 bytes: end of strings offset - # 4 bytes: start of string data offset + 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 strings + return - 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 data[0:4] != b'\x03\x00\x08\x00' and data[0:4] != b'\x00\x08\x00\x03': + return - if strings_end <= strings_start or strings_start < header_size: - return strings + strings = _extract_strings_fallback(data) - # String count is at offset header_size (right after header) - str_count_offset = header_size - if str_count_offset + 4 > len(data): - return strings + for s in strings: + if s and "." in s and not s.startswith("android"): + 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 - 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 + for s in strings: + if s and re.match(r'^\d+\.\d+', s): + if metadata["version_name"] is None: + metadata["version_name"] = s + break except Exception as e: - print(f"Error extracting strings: {e}") + print(f"Error in fallback manifest parsing: {e}") + +def _extract_strings_fallback(data: bytes) -> List[str]: + """Extract printable strings >= 4 chars from binary data.""" + strings = [] + current = [] + for byte in data: + if 32 <= byte < 127: + current.append(chr(byte)) + else: + if len(current) >= 4: + strings.append(''.join(current)) + current = [] + if len(current) >= 4: + strings.append(''.join(current)) return strings