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
This commit is contained in:
Jarian
parent
6c9d54b4ab
commit
10fbfb5c5a
@ -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> =
|
||||
|
||||
@ -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"):
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user