from __future__ import annotations import hashlib import io import json import os import re import shutil import subprocess import time import uuid import zipfile from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable from urllib.parse import quote from xml.etree import ElementTree from sqlalchemy.orm import Session from .database import get_db from .models import SoftwarePackage from .software_packages import build_generated_detection_script, build_generated_install_script, build_generated_uninstall_script, normalize_package_manifest, normalize_process_names, package_directory, unique_package_name, write_package_storage from .analyzer_profiles import ( apply_analysis_profiles, command_profile, detect_marker_profiles, marker_needles as profile_marker_needles, match_sfx_profiles, load_profiles, export_profile_bundle, import_profile_bundle, delete_imported_profile, set_profile_enabled, repository_index, install_repository_profile, apply_profile, installer_type_flag, REPOSITORY_URL as PROFILE_REPOSITORY_URL, ) from fastapi import Depends, File, Form, HTTPException, Request, UploadFile from fastapi.responses import RedirectResponse, StreamingResponse try: import pefile except Exception: pefile = None try: from cryptography.hazmat.primitives.serialization import pkcs7 from cryptography.x509.oid import ExtensionOID, ExtendedKeyUsageOID, NameOID except Exception: pkcs7 = None ExtensionOID = None ExtendedKeyUsageOID = None NameOID = None TEMP_ROOT = Path(os.getenv("SETUP_ANALYZER_TMP_DIR", "/tmp/assetmanager-setup-analyzer")) MAX_UPLOAD_MB = max(1, int(os.getenv("SETUP_ANALYZER_MAX_UPLOAD_MB", "4096"))) RETENTION_HOURS = max(1, int(os.getenv("SETUP_ANALYZER_RETENTION_HOURS", "24"))) MAX_EXTRACTED_MB = max(64, int(os.getenv("SETUP_ANALYZER_MAX_EXTRACTED_MB", "8192"))) MAX_EXTRACTED_FILES = max(100, int(os.getenv("SETUP_ANALYZER_MAX_EXTRACTED_FILES", "20000"))) MAX_SFX_DEPTH = max(0, min(int(os.getenv("SETUP_ANALYZER_MAX_SFX_DEPTH", "2")), 4)) ALLOWED_EXTENSIONS = {".exe", ".msi", ".msp", ".msix", ".appx", ".msu"} SCAN_CHUNK_BYTES = 4 * 1024 * 1024 SCAN_OVERLAP_BYTES = 2048 TEMP_ROOT.mkdir(parents=True, exist_ok=True) SFX_BINARY_SIGNATURES = { b"\x37\x7a\xbc\xaf\x27\x1c": "__7z_archive_signature__", b"Rar!\x1a\x07\x00": "__rar_archive_signature__", b"Rar!\x1a\x07\x01\x00": "__rar_archive_signature__", } SWITCH_MARKERS = [ b"/verysilent", b"/silent", b"/suppressmsgboxes", b"/norestart", b"/quiet", b"/qn", b"/passive", b"/s", b"--silent", b"--quiet", ] def _safe_filename(value: str | None) -> str: name = Path(value or "setup.bin").name name = re.sub(r"[^A-Za-z0-9._() +@-]", "_", name).strip(" .") return name[:180] or "setup.bin" def _human_size(size: int) -> str: value = float(size) for unit in ("B", "KB", "MB", "GB", "TB"): if value < 1024.0 or unit == "TB": return f"{int(value)} {unit}" if unit == "B" else f"{value:.1f} {unit}" value /= 1024.0 return f"{size} B" def _cleanup_old_files() -> None: threshold = time.time() - (RETENTION_HOURS * 3600) try: children = list(TEMP_ROOT.iterdir()) except OSError: return for child in children: try: if child.is_dir() and child.stat().st_mtime < threshold: shutil.rmtree(child, ignore_errors=True) except OSError: continue async def _save_upload(upload: UploadFile) -> tuple[str, Path, int, str]: _cleanup_old_files() filename = _safe_filename(upload.filename) extension = Path(filename).suffix.lower() if extension not in ALLOWED_EXTENSIONS: await upload.close() raise HTTPException(400, "Unsupported installer file type.") token = uuid.uuid4().hex job_dir = TEMP_ROOT / token job_dir.mkdir(mode=0o700, parents=True, exist_ok=False) target = job_dir / filename digest = hashlib.sha256() total = 0 limit = MAX_UPLOAD_MB * 1024 * 1024 try: with target.open("wb") as handle: while True: chunk = await upload.read(1024 * 1024) if not chunk: break total += len(chunk) if total > limit: raise HTTPException(413, f"Maximum upload size exceeded ({MAX_UPLOAD_MB} MB).") digest.update(chunk) handle.write(chunk) except Exception: shutil.rmtree(job_dir, ignore_errors=True) raise finally: await upload.close() if total == 0: shutil.rmtree(job_dir, ignore_errors=True) raise HTTPException(400, "The uploaded installer is empty.") return token, target, total, digest.hexdigest() def _analysis_file(token: str) -> tuple[Path, dict[str, Any]]: if not re.fullmatch(r"[0-9a-f]{32}", token or ""): raise HTTPException(400, "Invalid analysis token.") job_dir = TEMP_ROOT / token meta_file = job_dir / "analysis.json" if not meta_file.is_file(): raise HTTPException(404, "Analysis is no longer available.") try: meta = json.loads(meta_file.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise HTTPException(404, "Analysis metadata is not available.") from exc target = job_dir / _safe_filename(meta.get("filename")) if not target.is_file(): raise HTTPException(404, "Installer file is no longer available.") return target, meta def _scan_needles() -> dict[bytes, str]: result = profile_marker_needles() for marker in SWITCH_MARKERS: lower = marker.lower() normalized = lower.decode("ascii", errors="ignore") result[lower] = normalized result[normalized.encode("utf-16le")] = normalized return result def _scan_file_markers(path: Path) -> set[str]: found: set[str] = set() tail = b"" needles = _scan_needles() with path.open("rb") as handle: while True: chunk = handle.read(SCAN_CHUNK_BYTES) if not chunk: break combined = tail + chunk for raw, normalized in SFX_BINARY_SIGNATURES.items(): if normalized not in found and raw in combined: found.add(normalized) data = combined.lower() for raw, normalized in needles.items(): if normalized not in found and raw in data: found.add(normalized) tail = combined[-SCAN_OVERLAP_BYTES:] return found WRAPPER_INSTALLER_TYPES = { "inno", "nsis", "wix_burn", "installshield", "advanced_installer", "squirrel", "zip_sfx", "7zip_sfx", "winrar_sfx", } def _normalize_architecture(value: str) -> str: text = str(value or "").strip().casefold() if text in {"x64", "amd64", "x86_64", "x86-64", "win64", "64-bit", "64bit"}: return "x64" if text in {"x86", "i386", "i486", "i586", "i686", "win32", "32-bit", "32bit"}: return "x86" if text in {"arm64", "aarch64"}: return "arm64" if text in {"arm", "arm32"}: return "arm" return str(value or "").strip() def _filename_architecture_hint(filename: str) -> str: """Return only explicit architecture hints from a package filename. Installer EXE launchers are frequently 32-bit even when they deploy a 64-bit application. Therefore x86/x64 tokens in a vendor package filename are a better target-architecture signal than the PE machine type of a known setup bootstrapper. """ name = str(filename or "") stem = Path(name).stem.casefold() # Some vendor filenames append architecture directly to a version; combined # packages may also use tokens such as x32_64. if re.search(r"(?i)(?:x32[_-]?64|x64[_-]?32)$", stem): return "x86+x64" for architecture, suffixes in ( ("arm64", ("arm64", "aarch64")), ("x64", ("x64", "amd64", "win64")), ("x86", ("x86", "x32", "win32")), ): if any(stem.endswith(suffix) for suffix in suffixes): return architecture patterns = ( ("arm64", r"(?i)(?:^|[._+()\-\s])(?:arm64|aarch64)(?=$|[._+()\-\s])"), ("x64", r"(?i)(?:^|[._+()\-\s])(?:x64|amd64|x86[_-]?64|win64|64[-_ ]?bit)(?=$|[._+()\-\s])"), ("x86", r"(?i)(?:^|[._+()\-\s])(?:x86|x32|i[3-6]86|win32|32[-_ ]?bit)(?=$|[._+()\-\s])"), ) for architecture, pattern in patterns: if re.search(pattern, name): return architecture return "" def _resolve_target_architecture( installer_type: str, filename: str, pe_architecture: str = "", package_architecture: str = "", ) -> dict[str, str]: """Separate target architecture from the executable launcher's PE type.""" launcher = _normalize_architecture(pe_architecture) package_value = _normalize_architecture(package_architecture) filename_hint = _filename_architecture_hint(filename) if package_value: return { "architecture": package_value, "architecture_source": "package_metadata", "launcher_architecture": launcher, } if filename_hint: return { "architecture": filename_hint, "architecture_source": "filename", "launcher_architecture": launcher, } installer_key = str(installer_type or "").strip().casefold() if installer_key in WRAPPER_INSTALLER_TYPES or installer_type_flag(installer_key, "wrapper", False): # A 64-bit/ARM64 launcher itself requires that architecture, so it is a # useful fallback. A 32-bit launcher is *not* evidence that the payload is # x86: NSIS/Inno and other bootstrapper stubs commonly stay PE32 for x64 # products. if launcher in {"x64", "arm64"}: return { "architecture": launcher, "architecture_source": "launcher_requirement", "launcher_architecture": launcher, } return { "architecture": "", "architecture_source": "", "launcher_architecture": launcher, } return { "architecture": launcher, "architecture_source": "pe_machine" if launcher else "", "launcher_architecture": launcher, } def _filename_version_hint(filename: str) -> str: """Return a conservative dotted version token from an installer filename. This is intentionally generic. Product/vendor-specific compact version encodings are left to analyzer profiles. Common dotted version tokens can still provide useful metadata when the setup launcher itself has no VERSIONINFO resource. """ stem = Path(str(filename or "")).stem matches = list(re.finditer(r"(?i)(?:^|[._+()\-\s])v?(?P\d{1,4}(?:\.\d{1,4}){1,3})(?=$|[._+()\-\s])", stem)) if not matches: return "" # Prefer the most specific candidate (more components), then the first one. matches.sort(key=lambda item: (-item.group("version").count("."), item.start())) return matches[0].group("version") def _authenticode_metadata(path: Path) -> dict[str, Any]: """Read Authenticode certificate metadata without executing the file. The PE security directory is a file offset (not an RVA). We only use the certificate publisher as a fallback when installer metadata does not expose a manufacturer. A code-signing end-entity certificate is preferred over CA and timestamp certificates contained in the same PKCS#7 structure. """ result: dict[str, Any] = { "signature_present": None, "signature_publisher": "", "signature_subject": "", } try: with path.open("rb") as handle: header = handle.read(4096) if len(header) < 0x40 or header[:2] != b"MZ": return result pe_offset = int.from_bytes(header[0x3C:0x40], "little", signed=False) if pe_offset < 0 or pe_offset > 16 * 1024 * 1024: return result minimum = pe_offset + 4 + 20 + 2 if len(header) < minimum: handle.seek(0) header = handle.read(minimum + 256) if header[pe_offset:pe_offset + 4] != b"PE\x00\x00": return result optional_offset = pe_offset + 4 + 20 magic = int.from_bytes(header[optional_offset:optional_offset + 2], "little", signed=False) if magic == 0x10B: data_directory_offset = optional_offset + 96 elif magic == 0x20B: data_directory_offset = optional_offset + 112 else: return result security_entry_offset = data_directory_offset + (4 * 8) need = security_entry_offset + 8 if len(header) < need: handle.seek(0) header = handle.read(need) certificate_offset = int.from_bytes(header[security_entry_offset:security_entry_offset + 4], "little", signed=False) certificate_size = int.from_bytes(header[security_entry_offset + 4:security_entry_offset + 8], "little", signed=False) if not certificate_offset or certificate_size < 8: result["signature_present"] = False return result file_size = path.stat().st_size if certificate_offset >= file_size or certificate_offset + certificate_size > file_size: return result result["signature_present"] = True if pkcs7 is None: return result handle.seek(certificate_offset) certificate_table = handle.read(certificate_size) except (OSError, ValueError): return result certificates = [] position = 0 while position + 8 <= len(certificate_table): length = int.from_bytes(certificate_table[position:position + 4], "little", signed=False) certificate_type = int.from_bytes(certificate_table[position + 6:position + 8], "little", signed=False) if length < 8 or position + length > len(certificate_table): break if certificate_type == 0x0002: blob = certificate_table[position + 8:position + length] try: import warnings with warnings.catch_warnings(): warnings.simplefilter("ignore") certificates.extend(pkcs7.load_der_pkcs7_certificates(blob)) except Exception: pass position += (length + 7) & ~7 def certificate_score(certificate: Any) -> int: score = 0 try: constraints = certificate.extensions.get_extension_for_oid(ExtensionOID.BASIC_CONSTRAINTS).value score += -80 if constraints.ca else 40 except Exception: pass try: usage = certificate.extensions.get_extension_for_oid(ExtensionOID.EXTENDED_KEY_USAGE).value if ExtendedKeyUsageOID.CODE_SIGNING in usage: score += 120 if ExtendedKeyUsageOID.TIME_STAMPING in usage: score -= 100 except Exception: pass try: if certificate.subject != certificate.issuer: score += 10 except Exception: pass return score if not certificates: return result certificate = max(certificates, key=certificate_score) try: organizations = certificate.subject.get_attributes_for_oid(NameOID.ORGANIZATION_NAME) common_names = certificate.subject.get_attributes_for_oid(NameOID.COMMON_NAME) publisher = organizations[0].value if organizations else (common_names[0].value if common_names else "") result["signature_publisher"] = str(publisher or "").strip() result["signature_subject"] = certificate.subject.rfc4514_string() except Exception: pass return result def _pe_metadata(path: Path) -> dict[str, Any]: authenticode = _authenticode_metadata(path) result: dict[str, Any] = { "architecture": "", "company_name": "", "product_name": "", "product_version": "", "file_version": "", "original_filename": "", "signature_present": authenticode.get("signature_present"), "signature_publisher": authenticode.get("signature_publisher", ""), "signature_subject": authenticode.get("signature_subject", ""), } if pefile is None: return result try: pe = pefile.PE(str(path), fast_load=True) result["architecture"] = { 0x014C: "x86", 0x8664: "x64", 0xAA64: "arm64", }.get(int(pe.FILE_HEADER.Machine), hex(int(pe.FILE_HEADER.Machine))) security_index = pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_SECURITY"] security = pe.OPTIONAL_HEADER.DATA_DIRECTORY[security_index] result["signature_present"] = bool(security.VirtualAddress and security.Size) resource_index = pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_RESOURCE"] pe.parse_data_directories(directories=[resource_index]) values: dict[str, str] = {} for file_info in getattr(pe, "FileInfo", []) or []: items = file_info if isinstance(file_info, list) else [file_info] for item in items: key = getattr(item, "Key", b"") if isinstance(key, bytes): key = key.decode(errors="ignore") if key != "StringFileInfo": continue for string_table in getattr(item, "StringTable", []) or []: for raw_key, raw_value in (getattr(string_table, "entries", {}) or {}).items(): k = raw_key.decode(errors="ignore") if isinstance(raw_key, bytes) else str(raw_key) v = raw_value.decode(errors="ignore") if isinstance(raw_value, bytes) else str(raw_value) values[k] = v.strip() result["company_name"] = values.get("CompanyName", "") result["product_name"] = values.get("ProductName", "") result["product_version"] = values.get("ProductVersion", "") result["file_version"] = values.get("FileVersion", "") result["original_filename"] = values.get("OriginalFilename", "") pe.close() except Exception: return result return result def _run_parser(command: list[str], timeout: int = 20) -> str: try: completed = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", timeout=timeout, check=False, ) except (OSError, subprocess.SubprocessError): return "" return completed.stdout if completed.returncode == 0 else "" def _msi_properties(path: Path) -> dict[str, str]: if shutil.which("msiinfo") is None: return {} output = _run_parser(["msiinfo", "export", str(path), "Property"]) wanted = {"ProductName", "ProductVersion", "Manufacturer", "ProductCode", "UpgradeCode", "ALLUSERS"} result: dict[str, str] = {} for line in output.splitlines(): parts = line.split(" ") if len(parts) >= 2 and parts[0].strip() in wanted: result[parts[0].strip()] = parts[1].strip() return result def _msix_metadata(path: Path) -> dict[str, str]: result: dict[str, str] = {} try: with zipfile.ZipFile(path, "r") as archive: manifest_name = next((name for name in archive.namelist() if name.lower().endswith("appxmanifest.xml")), "") if not manifest_name: return result root = ElementTree.fromstring(archive.read(manifest_name)) identity = next((node for node in root.iter() if node.tag.endswith("Identity")), None) properties = next((node for node in root.iter() if node.tag.endswith("Properties")), None) if identity is not None: result["identity_name"] = identity.attrib.get("Name", "") result["publisher"] = identity.attrib.get("Publisher", "") result["version"] = identity.attrib.get("Version", "") result["architecture"] = identity.attrib.get("ProcessorArchitecture", "") if properties is not None: for node in properties: if node.tag.endswith("DisplayName") and node.text: result["display_name"] = node.text.strip() break except Exception: return {} return result def _detect_exe(found_markers: set[str]) -> tuple[str, str, int, list[str], list[dict[str, Any]]]: candidates = detect_marker_profiles(found_markers) if not candidates: return "unknown_exe", "Unknown EXE installer", 25, ["setup_analyzer.signal.exe"], [] primary = candidates[0] return ( str(primary["key"]), str(primary["label"]), max(55, int(primary["confidence"])), list(primary.get("signals") or []), candidates, ) def _command_defaults(installer_type: str, filename: str, product_code: str, product_name: str) -> dict[str, Any]: quoted = f'"{filename}"' result: dict[str, Any] = { "install_arguments": "", "install_command": quoted, "alternative_install_arguments": "", "alternative_install_command": "", "uninstall_command": "", "success_codes": [0], "reboot_codes": [], "detect_method": "registry_display_name" if product_name else "manual", "command_confidence": "none", "warning_keys": [], } if installer_type == "msi": result.update( install_arguments="/qn /norestart", install_command=f"msiexec.exe /i {quoted} /qn /norestart", success_codes=[0, 1641, 3010], reboot_codes=[1641, 3010], detect_method="msi_product_code" if product_code else "registry_display_name", command_confidence="high", ) if product_code: result["uninstall_command"] = f'msiexec.exe /x "{product_code}" /qn /norestart' elif installer_type == "msp": result.update( install_arguments="/qn /norestart", install_command=f"msiexec.exe /p {quoted} /qn /norestart", success_codes=[0, 1641, 3010], reboot_codes=[1641, 3010], command_confidence="high", ) elif installer_type == "msu": result.update( install_arguments="/quiet /norestart", install_command=f"wusa.exe {quoted} /quiet /norestart", success_codes=[0, 3010, 2359302], reboot_codes=[3010], command_confidence="high", ) elif command_profile(installer_type): profile_command = command_profile(installer_type) args = str(profile_command.get("install_arguments") or "").strip() alt_args = str(profile_command.get("alternative_install_arguments") or "").strip() if args: result["install_arguments"] = args result["install_command"] = f"{quoted} {args}" if alt_args: result["alternative_install_arguments"] = alt_args result["alternative_install_command"] = f"{quoted} {alt_args}" for key in ("success_codes", "reboot_codes", "detect_method", "command_confidence", "uninstall_command"): if key in profile_command: result[key] = profile_command[key] for warning in profile_command.get("warning_keys") or []: if warning not in result["warning_keys"]: result["warning_keys"].append(warning) elif installer_type in {"msix", "appx"}: result.update( install_arguments="", install_command=f"Add-AppxPackage -Path {quoted}", success_codes=[0], detect_method="appx_package", command_confidence="medium", ) result["warning_keys"].append("setup_analyzer.warning.appx_context") elif installer_type in {"zip_sfx", "7zip_sfx", "winrar_sfx"}: result["warning_keys"].append("setup_analyzer.warning.sfx") else: result["warning_keys"].append("setup_analyzer.warning.unknown") return result def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _archive_tool_status() -> dict[str, Any]: sevenzip = shutil.which("7zz") or shutil.which("7z") unar = shutil.which("unar") lsar = shutil.which("lsar") return { "sevenzip": sevenzip or "", "unar": unar or "", "lsar": lsar or "", "sevenzip_available": bool(sevenzip), "unar_available": bool(unar and lsar), } def _safe_archive_member(value: str) -> str: name = str(value or "").replace("\\", "/").strip() while name.startswith("./"): name = name[2:] if not name or name.startswith("/") or re.match(r"^[A-Za-z]:", name): raise ValueError("unsafe archive member path") parts = [part for part in name.split("/") if part not in {"", "."}] if not parts or any(part == ".." for part in parts): raise ValueError("unsafe archive member path") return "/".join(parts) def _validate_archive_listing(entries: list[tuple[str, int]], max_files: int, max_bytes: int) -> tuple[int, int]: file_count = 0 total_size = 0 for raw_name, raw_size in entries: _safe_archive_member(raw_name) file_count += 1 if file_count > max_files: raise ValueError("archive file count limit exceeded") try: size = max(0, int(raw_size or 0)) except (TypeError, ValueError): size = 0 total_size += size if total_size > max_bytes: raise ValueError("archive extracted size limit exceeded") return file_count, total_size def _zip_listing(path: Path) -> list[tuple[str, int]]: entries: list[tuple[str, int]] = [] with zipfile.ZipFile(path, "r") as archive: for info in archive.infolist(): if info.is_dir(): continue entries.append((info.filename, int(info.file_size or 0))) return entries def _extract_zip_safely(path: Path, destination: Path, max_files: int, max_bytes: int) -> dict[str, Any]: entries = _zip_listing(path) file_count, total_size = _validate_archive_listing(entries, max_files, max_bytes) destination.mkdir(parents=True, exist_ok=False) root = destination.resolve() with zipfile.ZipFile(path, "r") as archive: for info in archive.infolist(): if info.is_dir(): continue relative = _safe_archive_member(info.filename) target = (destination / relative).resolve() if root != target and root not in target.parents: raise ValueError("archive member escapes extraction directory") target.parent.mkdir(parents=True, exist_ok=True) with archive.open(info, "r") as source, target.open("wb") as output: shutil.copyfileobj(source, output, length=1024 * 1024) return {"extractor": "python-zipfile", "file_count": file_count, "extracted_bytes": total_size} def _sevenzip_listing(path: Path, executable: str) -> list[tuple[str, int]]: output = _run_parser([executable, "l", "-slt", str(path)], timeout=60) if not output: raise ValueError("7-Zip could not list the archive") entries: list[tuple[str, int]] = [] in_files = False current: dict[str, str] = {} for raw_line in output.splitlines() + [""]: line = raw_line.rstrip("\r\n") if line.startswith("----------"): in_files = True current = {} continue if not in_files: continue if not line: if current.get("Path") and current.get("Folder", "-") != "+": entries.append((current["Path"], int(current.get("Size") or 0))) current = {} continue if " = " in line: key, value = line.split(" = ", 1) current[key.strip()] = value.strip() if not entries: raise ValueError("7-Zip archive contains no files") return entries def _extract_with_sevenzip(path: Path, destination: Path, executable: str, max_files: int, max_bytes: int) -> dict[str, Any]: entries = _sevenzip_listing(path, executable) file_count, total_size = _validate_archive_listing(entries, max_files, max_bytes) destination.mkdir(parents=True, exist_ok=False) try: completed = subprocess.run( [executable, "x", "-y", f"-o{destination}", str(path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", timeout=180, check=False, ) except (OSError, subprocess.SubprocessError) as exc: raise ValueError(f"7-Zip extraction failed: {exc}") from exc if completed.returncode != 0: raise ValueError("7-Zip extraction failed") return {"extractor": Path(executable).name, "file_count": file_count, "extracted_bytes": total_size} def _walk_lsar_entries(value: Any) -> list[tuple[str, int]]: entries: list[tuple[str, int]] = [] if isinstance(value, dict): name = value.get("XADFileName") if name and not bool(value.get("XADIsDirectory")): entries.append((str(name), int(value.get("XADFileSize") or 0))) for child in value.values(): if isinstance(child, (dict, list)): entries.extend(_walk_lsar_entries(child)) elif isinstance(value, list): for child in value: entries.extend(_walk_lsar_entries(child)) return entries def _unar_listing(path: Path, lsar_executable: str) -> list[tuple[str, int]]: output = _run_parser([lsar_executable, "-json", str(path)], timeout=60) if not output: raise ValueError("lsar could not list the archive") try: data = json.loads(output) except ValueError as exc: raise ValueError("lsar returned invalid archive metadata") from exc entries = _walk_lsar_entries(data) if not entries: raise ValueError("archive contains no files") return entries def _extract_with_unar(path: Path, destination: Path, unar_executable: str, lsar_executable: str, max_files: int, max_bytes: int) -> dict[str, Any]: entries = _unar_listing(path, lsar_executable) file_count, total_size = _validate_archive_listing(entries, max_files, max_bytes) destination.mkdir(parents=True, exist_ok=False) try: completed = subprocess.run( [unar_executable, "-f", "-D", "-o", str(destination), str(path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", timeout=180, check=False, ) except (OSError, subprocess.SubprocessError) as exc: raise ValueError(f"unar extraction failed: {exc}") from exc if completed.returncode != 0: raise ValueError("unar extraction failed") return {"extractor": Path(unar_executable).name, "file_count": file_count, "extracted_bytes": total_size} def _validate_extracted_tree(root: Path, max_files: int, max_bytes: int) -> tuple[int, int]: base = root.resolve() count = 0 total = 0 for item in root.rglob("*"): if item.is_symlink(): raise ValueError("symbolic links are not allowed in extracted payloads") if not item.is_file(): continue resolved = item.resolve() if base != resolved and base not in resolved.parents: raise ValueError("extracted file escapes extraction directory") count += 1 if count > max_files: raise ValueError("archive file count limit exceeded") total += item.stat().st_size if total > max_bytes: raise ValueError("archive extracted size limit exceeded") return count, total def _extract_sfx(path: Path, destination: Path, installer_type: str, max_files: int, max_bytes: int) -> dict[str, Any]: tools = _archive_tool_status() errors: list[str] = [] attempted = False if zipfile.is_zipfile(path): attempted = True try: result = _extract_zip_safely(path, destination, max_files, max_bytes) count, total = _validate_extracted_tree(destination, max_files, max_bytes) result.update(status="success", file_count=count, extracted_bytes=total) return result except Exception as exc: shutil.rmtree(destination, ignore_errors=True) errors.append(str(exc)) sevenzip = str(tools.get("sevenzip") or "") unar = str(tools.get("unar") or "") lsar = str(tools.get("lsar") or "") methods: list[str] = [] if installer_type == "winrar_sfx": if unar and lsar: methods.append("unar") if sevenzip: methods.append("7zip") else: if sevenzip: methods.append("7zip") if unar and lsar: methods.append("unar") for method in methods: attempted = True shutil.rmtree(destination, ignore_errors=True) try: if method == "7zip": result = _extract_with_sevenzip(path, destination, sevenzip, max_files, max_bytes) else: result = _extract_with_unar(path, destination, unar, lsar, max_files, max_bytes) count, total = _validate_extracted_tree(destination, max_files, max_bytes) result.update(status="success", file_count=count, extracted_bytes=total) return result except Exception as exc: errors.append(str(exc)) shutil.rmtree(destination, ignore_errors=True) if not attempted: return {"status": "tool_missing", "extractor": "", "file_count": 0, "extracted_bytes": 0, "error": "No SFX extraction tool is available."} return {"status": "failed", "extractor": "", "file_count": 0, "extracted_bytes": 0, "error": "; ".join(errors[-3:])[:1000]} def _candidate_score(analysis: dict[str, Any], relative_path: str) -> int: installer_type = str(analysis.get("installer_type") or "") base = { "msi": 850, "msp": 650, "msix": 700, "appx": 700, "msu": 500, "inno": 780, "nsis": 760, "wix_burn": 750, "advanced_installer": 720, "installshield": 680, "squirrel": 600, "zip_sfx": 350, "7zip_sfx": 350, "winrar_sfx": 350, "unknown_exe": 180, }.get(installer_type, 100) name = Path(relative_path).name.casefold() stem = Path(relative_path).stem.casefold() score = base + int(analysis.get("confidence") or 0) if stem in {"setup", "install", "installer"}: score += 260 elif any(token in stem for token in ("setup", "install", "installer")): score += 80 if any(token in name for token in ("uninstall", "unins000", "unins001", "remove")): score -= 1200 if any(token in name for token in ("vc_redist", "vcredist", "dotnet", "directx", "prereq", "prerequisite")): score -= 350 if analysis.get("product_name"): score += 30 if analysis.get("manufacturer"): score += 10 return score def _analyze_basic_file(path: Path) -> dict[str, Any]: filename = path.name extension = path.suffix.lower() with path.open("rb") as handle: magic = handle.read(8) found_markers = _scan_file_markers(path) pe = _pe_metadata(path) if extension == ".exe" or magic[:2] == b"MZ" else {} msi = _msi_properties(path) if extension in {".msi", ".msp"} else {} msix = _msix_metadata(path) if extension in {".msix", ".appx"} else {} installer_type = "unknown" installer_label = "Unknown package" confidence = 20 signal_keys = ["setup_analyzer.signal.unknown"] candidates: list[dict[str, Any]] = [] primary_profile_id = "" if extension == ".msi": installer_type, installer_label, confidence = "msi", "Windows Installer (MSI)", 99 signal_keys = ["setup_analyzer.signal.msi_extension"] elif extension == ".msp": installer_type, installer_label, confidence = "msp", "Windows Installer Patch (MSP)", 99 signal_keys = ["setup_analyzer.signal.msp_extension"] elif extension == ".msix": installer_type, installer_label, confidence = "msix", "MSIX", 99 signal_keys = ["setup_analyzer.signal.msix"] elif extension == ".appx": installer_type, installer_label, confidence = "appx", "AppX", 99 signal_keys = ["setup_analyzer.signal.appx"] elif extension == ".msu": installer_type, installer_label, confidence = "msu", "Windows Update Standalone Package (MSU)", 99 signal_keys = ["setup_analyzer.signal.msu"] elif extension == ".exe" or magic[:2] == b"MZ": installer_type, installer_label, confidence, signal_keys, candidates = _detect_exe(found_markers) if candidates: primary_profile_id = str(candidates[0].get("profile_id") or "") if installer_type == "unknown_exe" and zipfile.is_zipfile(path): installer_type, installer_label, confidence = "zip_sfx", "ZIP self-extracting archive", 85 signal_keys = ["setup_analyzer.signal.zip_sfx"] product_name = msi.get("ProductName") or msix.get("display_name") or msix.get("identity_name") or pe.get("product_name") or "" product_version = msi.get("ProductVersion") or msix.get("version") or pe.get("product_version") or pe.get("file_version") or _filename_version_hint(filename) or "" manufacturer = msi.get("Manufacturer") or msix.get("publisher") or pe.get("company_name") or pe.get("signature_publisher") or "" product_version_source = ( "package_metadata" if (msi.get("ProductVersion") or msix.get("version")) else "pe_version" if (pe.get("product_version") or pe.get("file_version")) else "filename" if product_version else "" ) manufacturer_source = ( "package_metadata" if (msi.get("Manufacturer") or msix.get("publisher")) else "pe_version" if pe.get("company_name") else "authenticode_signer" if manufacturer else "" ) architecture_info = _resolve_target_architecture( installer_type, filename, pe_architecture=pe.get("architecture", ""), package_architecture=msix.get("architecture", ""), ) product_code = msi.get("ProductCode", "") upgrade_code = msi.get("UpgradeCode", "") defaults = _command_defaults(installer_type, filename, product_code, product_name) warning_keys = list(defaults.pop("warning_keys", [])) if extension in {".msi", ".msp"} and not msi: warning_keys.append("setup_analyzer.warning.msiinfo") if (extension == ".exe" or magic[:2] == b"MZ") and pefile is None: warning_keys.append("setup_analyzer.warning.pefile") launcher_architecture = architecture_info["launcher_architecture"] architecture = architecture_info["architecture"] if (launcher_architecture and architecture and launcher_architecture != architecture and (installer_type in WRAPPER_INSTALLER_TYPES or installer_type_flag(installer_type, "wrapper", False))): warning_keys.append("setup_analyzer.warning.wrapper_architecture") embedded_switches = [marker.decode("ascii") for marker in SWITCH_MARKERS if marker.decode("ascii").lower() in found_markers] analysis = { "filename": filename, "extension": extension, "installer_type": installer_type, "installer_label": installer_label, "confidence": confidence, "candidates": candidates, "product_name": product_name, "product_version": product_version, "manufacturer": manufacturer, "product_version_source": product_version_source, "manufacturer_source": manufacturer_source, "signature_publisher": pe.get("signature_publisher", ""), "architecture": architecture, "launcher_architecture": launcher_architecture, "architecture_source": architecture_info["architecture_source"], "product_code": product_code, "upgrade_code": upgrade_code, "signature_present": pe.get("signature_present") if pe else None, "original_filename": pe.get("original_filename", "") if pe else "", "signal_keys": signal_keys, "embedded_switches": embedded_switches, "suppress_browser_default": False, "process_names_default": "", "start_application_default": False, "start_executable_default": "", "start_arguments_default": "", "warning_keys": warning_keys, **defaults, } if primary_profile_id: analysis = apply_profile(primary_profile_id, analysis, filename) analysis = apply_analysis_profiles(analysis, filename, found_markers) return analysis def _read_relaxed_ini(path: Path) -> dict[str, dict[str, str]]: """Read installer metadata INI files without executing or trusting them.""" raw = path.read_bytes() text = "" for encoding in ("utf-8-sig", "cp1252", "latin-1"): try: text = raw.decode(encoding) break except UnicodeDecodeError: continue sections: dict[str, dict[str, str]] = {} current = "" for raw_line in text.splitlines(): line = raw_line.strip() if not line or line.startswith((";", "#", "//")): continue if line.startswith("[") and line.endswith("]"): current = line[1:-1].strip().casefold() sections.setdefault(current, {}) continue if not current or "=" not in line: continue key, value = line.split("=", 1) sections.setdefault(current, {})[key.strip().casefold()] = value.strip() return sections def _analyze_sfx_recursive( archive_path: Path, archive_type: str, job_dir: Path, depth: int, budget: dict[str, int], sequence: list[int], ) -> dict[str, Any]: sequence[0] += 1 extract_dir = job_dir / f"sfx-extracted-{sequence[0]:03d}" remaining_files = max(1, MAX_EXTRACTED_FILES - budget.get("files", 0)) remaining_bytes = max(1, (MAX_EXTRACTED_MB * 1024 * 1024) - budget.get("bytes", 0)) result = _extract_sfx(archive_path, extract_dir, archive_type, remaining_files, remaining_bytes) result["depth"] = depth result["container_file"] = archive_path.name result["candidates"] = [] if result.get("status") != "success": return result budget["files"] = budget.get("files", 0) + int(result.get("file_count") or 0) budget["bytes"] = budget.get("bytes", 0) + int(result.get("extracted_bytes") or 0) root_rel = extract_dir.relative_to(job_dir).as_posix() result["source_root_rel"] = root_rel candidate_files = [ item for item in extract_dir.rglob("*") if item.is_file() and item.suffix.lower() in ALLOWED_EXTENSIONS ] candidate_files.sort(key=lambda item: (len(item.relative_to(extract_dir).parts), item.as_posix().casefold())) for candidate_path in candidate_files: relative_path = candidate_path.relative_to(extract_dir).as_posix() try: basic = _analyze_basic_file(candidate_path) except Exception: continue record = { "relative_path": relative_path, "source_root_rel": root_rel, "installer_type": basic.get("installer_type", ""), "installer_label": basic.get("installer_label", ""), "confidence": int(basic.get("confidence") or 0), "product_name": basic.get("product_name", ""), "product_version": basic.get("product_version", ""), "manufacturer": basic.get("manufacturer", ""), "architecture": basic.get("architecture", ""), "launcher_architecture": basic.get("launcher_architecture", ""), "architecture_source": basic.get("architecture_source", ""), "score": _candidate_score(basic, relative_path), "analysis": basic, } result["candidates"].append(record) if basic.get("installer_type") in {"zip_sfx", "7zip_sfx", "winrar_sfx"} and depth < MAX_SFX_DEPTH: nested = _analyze_sfx_recursive(candidate_path, str(basic.get("installer_type")), job_dir, depth + 1, budget, sequence) for nested_candidate in nested.get("candidates") or []: nested_candidate["score"] = int(nested_candidate.get("score") or 0) - ((depth + 1) * 15) result["candidates"].append(nested_candidate) result["candidates"].sort(key=lambda item: (int(item.get("score") or 0), int(item.get("confidence") or 0)), reverse=True) return result def _public_sfx_candidate(record: dict[str, Any]) -> dict[str, Any]: return { "relative_path": record.get("relative_path", ""), "installer_type": record.get("installer_type", ""), "installer_label": record.get("installer_label", ""), "confidence": record.get("confidence", 0), "product_name": record.get("product_name", ""), "product_version": record.get("product_version", ""), "manufacturer": record.get("manufacturer", ""), "architecture": record.get("architecture", ""), "launcher_architecture": record.get("launcher_architecture", ""), "architecture_source": record.get("architecture_source", ""), "score": record.get("score", 0), } def _create_payload_archive(target: Path, meta: dict[str, Any]) -> Path: root_rel = str(meta.get("deployment_payload_dir") or "").strip() if not root_rel: return target root = (target.parent / root_rel).resolve() job_root = target.parent.resolve() if job_root != root and job_root not in root.parents: raise HTTPException(400, "Invalid embedded payload directory.") if not root.is_dir(): raise HTTPException(404, "Embedded payload is no longer available.") archive_path = target.parent / "embedded-payload.zip" if archive_path.exists(): archive_path.unlink() count, total = _validate_extracted_tree(root, MAX_EXTRACTED_FILES, MAX_EXTRACTED_MB * 1024 * 1024) if count <= 0 or total <= 0: raise HTTPException(400, "Embedded payload is empty.") with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as archive: for item in sorted(root.rglob("*")): if item.is_file(): archive.write(item, arcname=item.relative_to(root).as_posix()) return archive_path def _deployment_source_file(target: Path, meta: dict[str, Any]) -> Path: if str(meta.get("deployment_source") or "direct") == "embedded_payload": return _create_payload_archive(target, meta) return target def analyze_file(path: Path, token: str, size: int, sha256: str) -> dict[str, Any]: base = _analyze_basic_file(path) analysis = dict(base) analysis.update({ "token": token, "filename": path.name, "size": size, "size_human": _human_size(size), "sha256": sha256, "analyzed_at": datetime.now(timezone.utc).isoformat(), "msiinfo_available": shutil.which("msiinfo") is not None, "pefile_available": pefile is not None, "archive_tools": _archive_tool_status(), "sfx_analysis": None, "deployment_source": "direct", "deployment_payload_dir": "", "embedded_installer_path": "", }) outer_type = str(base.get("installer_type") or "") if outer_type in {"zip_sfx", "7zip_sfx", "winrar_sfx"}: budget = {"files": 0, "bytes": 0} sequence = [0] sfx = _analyze_sfx_recursive(path, outer_type, path.parent, 0, budget, sequence) public_sfx = { "status": sfx.get("status", "failed"), "extractor": sfx.get("extractor", ""), "file_count": sfx.get("file_count", 0), "extracted_bytes": sfx.get("extracted_bytes", 0), "extracted_size_human": _human_size(int(sfx.get("extracted_bytes") or 0)), "error": sfx.get("error", ""), "container_type": outer_type, "container_label": base.get("installer_label", ""), "container_confidence": base.get("confidence", 0), "candidates": [_public_sfx_candidate(item) for item in (sfx.get("candidates") or [])[:20]], "selected": None, } analysis["sfx_analysis"] = public_sfx warning_keys = [key for key in analysis.get("warning_keys", []) if key != "setup_analyzer.warning.sfx"] selectable_candidates = [ item for item in (sfx.get("candidates") or []) if item.get("installer_type") not in {"zip_sfx", "7zip_sfx", "winrar_sfx"} ] vendor_profile = None if sfx.get("status") == "success": root_rel = str(sfx.get("source_root_rel") or "").strip() if root_rel: extract_root = (path.parent / root_rel).resolve() if extract_root.is_dir(): vendor_profile = match_sfx_profiles(extract_root, path, base) if sfx.get("status") == "success" and selectable_candidates: selected = selectable_candidates[0] selected_analysis = dict(selected.get("analysis") or {}) public_sfx["selected"] = _public_sfx_candidate(selected) analysis["outer_installer_type"] = outer_type analysis["outer_installer_label"] = base.get("installer_label", "") analysis["outer_confidence"] = base.get("confidence", 0) analysis["installer_type"] = selected_analysis.get("installer_type", outer_type) analysis["installer_label"] = selected_analysis.get("installer_label", base.get("installer_label", "")) analysis["confidence"] = selected_analysis.get("confidence", base.get("confidence", 0)) for key in ( "product_name", "product_version", "manufacturer", "architecture", "launcher_architecture", "architecture_source", "product_code", "upgrade_code", "signal_keys", "embedded_switches", "install_arguments", "install_command", "uninstall_command", "success_codes", "reboot_codes", "detect_method", "command_confidence", "suppress_browser_default", "process_names_default", "start_application_default", "start_executable_default", "start_arguments_default", ): if key in selected_analysis: analysis[key] = selected_analysis[key] for key in selected_analysis.get("warning_keys", []): if key not in warning_keys and key != "setup_analyzer.warning.sfx": warning_keys.append(key) warning_keys.append("setup_analyzer.warning.sfx_embedded_selected") analysis["deployment_source"] = "embedded_payload" analysis["deployment_payload_dir"] = selected.get("source_root_rel", "") analysis["embedded_installer_path"] = selected.get("relative_path", "") selected_command = str(analysis.get("install_command") or "") selected_name = str(selected_analysis.get("filename") or "") if selected_name and selected_command: analysis["install_command"] = selected_command.replace(f'"{selected_name}"', f'"{selected.get("relative_path", selected_name)}"', 1) elif vendor_profile: analysis["outer_installer_type"] = outer_type analysis["outer_installer_label"] = base.get("installer_label", "") analysis["outer_confidence"] = base.get("confidence", 0) for key, value in vendor_profile.items(): analysis[key] = value analysis["deployment_source"] = "direct" analysis["deployment_payload_dir"] = "" analysis["embedded_installer_path"] = "" analysis["warning_keys"] = [ key for key in warning_keys if key not in { "setup_analyzer.warning.sfx", "setup_analyzer.warning.sfx_no_installer", } ] else: if sfx.get("status") == "tool_missing": warning_keys.append("setup_analyzer.warning.sfx_tool_missing") elif sfx.get("status") == "success": warning_keys.append("setup_analyzer.warning.sfx_no_installer") else: warning_keys.append("setup_analyzer.warning.sfx_extract_failed") analysis["warning_keys"] = warning_keys (path.parent / "analysis.json").write_text(json.dumps(analysis, ensure_ascii=True, indent=2), encoding="utf-8") return analysis def _form_value(value: str | None, fallback: str = "") -> str: return (value if value is not None else fallback).strip() def _parse_codes(value: str, fallback: list[int]) -> list[int]: result: list[int] = [] for item in re.split(r"[,; ]+", value.strip()): if not item: continue try: number = int(item) except ValueError: continue if number not in result: result.append(number) return result or list(fallback) def _ps_quote(value: str) -> str: return "'" + value.replace("'", "''") + "'" def _powershell_install( filename: str, installer_type: str, install_arguments: str, success_codes: list[int], reboot_codes: list[int], timeout_seconds: int = 600, suppress_browser: bool = False, ) -> str: success = ", ".join(str(code) for code in success_codes) reboot = ", ".join(str(code) for code in reboot_codes) or "-999999" timeout_seconds = max(30, min(int(timeout_seconds or 600), 86400)) lines = [ "$ErrorActionPreference = 'Stop'", "Set-StrictMode -Version Latest", "", "$packageDir = $PSScriptRoot", f"$installer = Join-Path $packageDir {_ps_quote(filename)}", f"$arguments = {_ps_quote(install_arguments)}", f"$successCodes = @({success})", f"$rebootCodes = @({reboot})", f"$timeoutSeconds = {timeout_seconds}", "$suppressBrowser = $" + ("true" if suppress_browser else "false"), "", "function Stop-InstallerBrowserDescendants([int]$RootPid) {", "\tif (-not $suppressBrowser) { return }", "\t$browserNames = @('msedge.exe','chrome.exe','firefox.exe','brave.exe','opera.exe','iexplore.exe')", "\ttry { $rows = @(Get-CimInstance Win32_Process -ErrorAction Stop) } catch { return }", "\t$descendants = @($RootPid)", "\t$changed = $true", "\twhile ($changed) {", "\t\t$changed = $false", "\t\tforeach ($row in $rows) {", "\t\t\t$pidValue = [int]$row.ProcessId", "\t\t\t$parentValue = [int]$row.ParentProcessId", "\t\t\tif (($descendants -contains $parentValue) -and ($descendants -notcontains $pidValue)) {", "\t\t\t\t$descendants += $pidValue", "\t\t\t\t$changed = $true", "\t\t\t}", "\t\t}", "\t}", "\tforeach ($row in $rows) {", "\t\t$pidValue = [int]$row.ProcessId", "\t\t$nameValue = ([string]$row.Name).ToLowerInvariant()", "\t\tif (($pidValue -ne $RootPid) -and ($descendants -contains $pidValue) -and ($browserNames -contains $nameValue)) {", "\t\t\ttry { Stop-Process -Id $pidValue -Force -ErrorAction SilentlyContinue } catch {}", "\t\t}", "\t}", "}", "", "function Wait-InstallerProcess([System.Diagnostics.Process]$Process) {", "\ttry {", "\t\tWait-Process -Id $Process.Id -Timeout $timeoutSeconds -ErrorAction Stop", "\t} catch {", "\t\ttry { Stop-Process -Id $Process.Id -Force -ErrorAction SilentlyContinue } catch {}", "\t\tWrite-Error (\"Installer timeout after $timeoutSeconds seconds.\")", "\t\texit 1460", "\t}", "\tif ($suppressBrowser) {", "\t\tStart-Sleep -Milliseconds 1500", "\t\tStop-InstallerBrowserDescendants -RootPid $Process.Id", "\t}", "\t$Process.Refresh()", "\treturn [int]$Process.ExitCode", "}", "", "if (-not (Test-Path -LiteralPath $installer)) {", '\tWrite-Error "Installer not found: $installer"', "\texit 2", "}", "", ] if installer_type == "msi": lines.extend([ "$processArguments = '/i \"' + $installer + '\" ' + $arguments", "$process = Start-Process -FilePath 'msiexec.exe' -ArgumentList $processArguments -PassThru -NoNewWindow", ]) elif installer_type == "msp": lines.extend([ "$processArguments = '/p \"' + $installer + '\" ' + $arguments", "$process = Start-Process -FilePath 'msiexec.exe' -ArgumentList $processArguments -PassThru -NoNewWindow", ]) elif installer_type == "msu": lines.extend([ "$processArguments = '\"' + $installer + '\" ' + $arguments", "$process = Start-Process -FilePath 'wusa.exe' -ArgumentList $processArguments -PassThru -NoNewWindow", ]) elif installer_type in {"msix", "appx"}: lines.extend(["Add-AppxPackage -Path $installer -ErrorAction Stop", "exit 0"]) return "\n".join(lines) + "\n" else: lines.append("$process = Start-Process -FilePath $installer -ArgumentList $arguments -PassThru -NoNewWindow") lines.extend([ "$exitCode = Wait-InstallerProcess -Process $process", 'Write-Output "Installer exit code: $exitCode"', "if ($successCodes -notcontains $exitCode) { exit $exitCode }", "if ($rebootCodes -contains $exitCode) { exit 3010 }", "exit 0", ]) return "\n".join(lines) + "\n" def _powershell_detect(product_code: str, product_name: str, installer_type: str) -> str: if product_code: return f"""$ErrorActionPreference = 'SilentlyContinue'\n$productCode = {_ps_quote(product_code)}\n$paths = @(\n \"HKLM:\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\$productCode\",\n \"HKLM:\\SOFTWARE\\WOW6432Node\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\$productCode\"\n)\nif ($paths | Where-Object {{ Test-Path -LiteralPath $_ }}) {{ exit 0 }}\nexit 1\n""" if installer_type in {"msix", "appx"} and product_name: return f"""$ErrorActionPreference = 'SilentlyContinue'\n$name = {_ps_quote(product_name)}\n$package = Get-AppxPackage -AllUsers | Where-Object {{ $_.Name -eq $name -or $_.PackageFullName -like \"$name*\" }} | Select-Object -First 1\nif ($null -ne $package) {{ exit 0 }}\nexit 1\n""" if product_name: return f"""$ErrorActionPreference = 'SilentlyContinue'\n$displayName = {_ps_quote(product_name)}\n$roots = @(\n 'HKLM:\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*',\n 'HKLM:\\SOFTWARE\\WOW6432Node\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'\n)\n$match = Get-ItemProperty -Path $roots -ErrorAction SilentlyContinue | Where-Object {{ $_.DisplayName -eq $displayName }} | Select-Object -First 1\nif ($null -ne $match) {{ exit 0 }}\nexit 1\n""" return "Write-Output 'No automatic detection rule is available for this package.'\nexit 2\n" def _powershell_uninstall(product_code: str, product_name: str, installer_type: str) -> str: if product_code: return f"""$ErrorActionPreference = 'Stop'\n$productCode = {_ps_quote(product_code)}\n$arguments = '/x \"' + $productCode + '\" /qn /norestart'\n$process = Start-Process -FilePath 'msiexec.exe' -ArgumentList $arguments -Wait -PassThru -NoNewWindow\nif (@(0, 1641, 3010) -notcontains [int]$process.ExitCode) {{ exit [int]$process.ExitCode }}\nif (@(1641, 3010) -contains [int]$process.ExitCode) {{ exit 3010 }}\nexit 0\n""" if installer_type in {"msix", "appx"} and product_name: return f"""$ErrorActionPreference = 'Stop'\n$name = {_ps_quote(product_name)}\n$packages = Get-AppxPackage -AllUsers | Where-Object {{ $_.Name -eq $name -or $_.PackageFullName -like \"$name*\" }}\nforeach ($package in $packages) {{ Remove-AppxPackage -Package $package.PackageFullName -AllUsers -ErrorAction Stop }}\nexit 0\n""" if product_name: return f"""$ErrorActionPreference = 'Stop'\n$displayName = {_ps_quote(product_name)}\n$roots = @(\n 'HKLM:\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*',\n 'HKLM:\\SOFTWARE\\WOW6432Node\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'\n)\n$entry = Get-ItemProperty -Path $roots -ErrorAction SilentlyContinue | Where-Object {{ $_.DisplayName -eq $displayName }} | Select-Object -First 1\nif ($null -eq $entry) {{ exit 0 }}\n$command = $entry.QuietUninstallString\nif ([string]::IsNullOrWhiteSpace($command)) {{ $command = $entry.UninstallString }}\nif ([string]::IsNullOrWhiteSpace($command)) {{ Write-Error 'No uninstall command was found in the registry.'; exit 3 }}\n$process = Start-Process -FilePath 'cmd.exe' -ArgumentList @('/d', '/s', '/c', $command) -Wait -PassThru -NoNewWindow\nif (@(0, 1641, 3010) -notcontains [int]$process.ExitCode) {{ exit [int]$process.ExitCode }}\nif (@(1641, 3010) -contains [int]$process.ExitCode) {{ exit 3010 }}\nexit 0\n""" return "Write-Error 'No automatic uninstall rule is available for this package.'\nexit 2\n" def _safe_package_name(name: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", name.strip()).strip("-") return cleaned[:80] or "software-package" def _export_values( meta: dict[str, Any], product_name: str, product_version: str, manufacturer: str, architecture: str, install_arguments: str, timeout_seconds: int, run_as: str, success_codes: str, reboot_codes: str, suppress_browser: bool = False, process_names: str = "", start_application: bool = False, start_executable: str = "", start_arguments: str = "", start_only_if_user_logged_on: bool = True, start_fail_job_on_error: bool = False, ) -> dict[str, Any]: return { "product_name": _form_value(product_name, meta.get("product_name", "")), "product_version": _form_value(product_version, meta.get("product_version", "")), "manufacturer": _form_value(manufacturer, meta.get("manufacturer", "")), "architecture": _form_value(architecture, meta.get("architecture", "")), "install_arguments": _form_value(install_arguments, meta.get("install_arguments", "")), "timeout_seconds": max(30, min(int(timeout_seconds), 86400)), "run_as": run_as if run_as in {"system", "user"} else "system", "success_codes": _parse_codes(success_codes, meta.get("success_codes") or [0]), "reboot_codes": _parse_codes(reboot_codes, meta.get("reboot_codes") or []), "suppress_browser": bool(suppress_browser), "process_names": normalize_process_names(process_names), "start_application": bool(start_application), "start_executable": _form_value(start_executable, meta.get("start_executable_default", ""))[:1024], "start_arguments": str(start_arguments or "").strip()[:2048], "start_only_if_user_logged_on": bool(start_only_if_user_logged_on), "start_fail_job_on_error": bool(start_fail_job_on_error), } def _build_package_manifest( target: Path, meta: dict[str, Any], values: dict[str, Any], ) -> dict[str, Any]: detection_method = str(meta.get("detect_method") or "manual") if meta.get("product_code"): detection_method = "msi_product_code" elif meta.get("installer_type") in {"msix", "appx"} and values.get("product_name"): detection_method = "appx_package" elif values.get("product_name"): detection_method = "registry_display_name" return { "schema": "assetmanager-software-package-v1", "name": values["product_name"] or target.stem, "version": values["product_version"], "vendor": values["manufacturer"], "architecture": values["architecture"], "platform": "windows", "installer_type": meta.get("installer_type", ""), "installer_file": target.name, "install": { "script": "install.ps1", "source_mode": "embedded_archive" if meta.get("deployment_source") == "embedded_payload" else "direct", "embedded_installer": meta.get("embedded_installer_path", "") if meta.get("deployment_source") == "embedded_payload" else "", "arguments": values["install_arguments"], "timeout_seconds": values["timeout_seconds"], "run_as": values["run_as"], "success_codes": values["success_codes"], "reboot_codes": values["reboot_codes"], "suppress_browser": values["suppress_browser"], }, "uninstall": {"script": "uninstall.ps1"}, "process_control": { "process_names": values["process_names"], "grace_seconds": 5, "force_close": True, }, "post_install": { "start_application": values["start_application"], "executable": values["start_executable"], "arguments": values["start_arguments"], "only_if_user_logged_on": values["start_only_if_user_logged_on"], "fail_job_on_error": values["start_fail_job_on_error"], }, "detection": { "script": "detect.ps1", "method": detection_method, "product_code": meta.get("product_code", ""), "display_name": values["product_name"], "display_version": values["product_version"], "publisher": values["manufacturer"], }, "analysis": { "sha256": meta.get("sha256", ""), "confidence": meta.get("confidence", 0), "command_confidence": meta.get("command_confidence", "none"), "container_type": meta.get("outer_installer_type", ""), "embedded_installer": meta.get("embedded_installer_path", ""), "profile_id": meta.get("profile_id", ""), "profile_name": meta.get("profile_name", ""), "profile_version": meta.get("profile_version", ""), "profile_source": meta.get("profile_source", ""), }, } def register_setup_analyzer(app: Any, templates: Any, require_admin: Callable[[Request], None]) -> None: @app.get("/software/setup-analyzer") def setup_analyzer_page(request: Request): require_admin(request) return templates.TemplateResponse("setup_analyzer.html", {"request": request, "analysis": None, "max_upload_mb": MAX_UPLOAD_MB, "max_extracted_mb": MAX_EXTRACTED_MB, "max_extracted_files": MAX_EXTRACTED_FILES, "max_sfx_depth": MAX_SFX_DEPTH}) @app.get("/software/setup-analyzer/profiles") def setup_analyzer_profiles_page(request: Request, repository: int = 0): require_admin(request) repository_data: dict[str, Any] = {"configured": bool(PROFILE_REPOSITORY_URL), "url": PROFILE_REPOSITORY_URL, "profiles": []} repository_error = "" if repository and PROFILE_REPOSITORY_URL: try: repository_data = repository_index() except Exception as exc: repository_error = str(exc) return templates.TemplateResponse("setup_analyzer_profiles.html", { "request": request, "profiles": load_profiles(include_disabled=True), "repository": repository_data, "repository_error": repository_error, }) @app.post("/software/setup-analyzer/profiles/import") async def setup_analyzer_profile_import(request: Request, profile_file: UploadFile = File(...), source: str = Form("community")): require_admin(request) try: data = await profile_file.read(2 * 1024 * 1024 + 1) profile = import_profile_bundle(data, source=source if source in {"community", "local"} else "community") except Exception as exc: return RedirectResponse("/software/setup-analyzer/profiles?toast_error=" + quote(str(exc)), status_code=303) finally: await profile_file.close() return RedirectResponse("/software/setup-analyzer/profiles?toast_success=" + quote(f"Analyzer profile {profile.get('name', profile.get('id', ''))} imported."), status_code=303) @app.get("/software/setup-analyzer/profiles/{profile_id}/export") def setup_analyzer_profile_export(profile_id: str, request: Request): require_admin(request) try: data = export_profile_bundle(profile_id) except Exception as exc: raise HTTPException(404, str(exc)) from exc safe = re.sub(r"[^A-Za-z0-9._-]+", "-", profile_id).strip("-.") or "analyzer-profile" return StreamingResponse(io.BytesIO(data), media_type="application/zip", headers={"Content-Disposition": f'attachment; filename="{safe}.amprofile"'}) @app.post("/software/setup-analyzer/profiles/{profile_id}/toggle") async def setup_analyzer_profile_toggle(profile_id: str, request: Request): require_admin(request) form = await request.form() enabled = str(form.get("enabled") or "").strip().lower() in {"1", "true", "yes", "on"} try: set_profile_enabled(profile_id, enabled) except Exception as exc: return RedirectResponse("/software/setup-analyzer/profiles?toast_error=" + quote(str(exc)), status_code=303) return RedirectResponse("/software/setup-analyzer/profiles", status_code=303) @app.post("/software/setup-analyzer/profiles/{profile_id}/delete") def setup_analyzer_profile_delete(profile_id: str, request: Request): require_admin(request) try: if not delete_imported_profile(profile_id): raise ValueError("System profiles cannot be deleted.") except Exception as exc: return RedirectResponse("/software/setup-analyzer/profiles?toast_error=" + quote(str(exc)), status_code=303) return RedirectResponse("/software/setup-analyzer/profiles", status_code=303) @app.post("/software/setup-analyzer/profiles/repository/install") async def setup_analyzer_repository_install(request: Request): require_admin(request) form = await request.form() profile_id = str(form.get("profile_id") or "") try: profile = install_repository_profile(profile_id) except Exception as exc: return RedirectResponse("/software/setup-analyzer/profiles?repository=1&toast_error=" + quote(str(exc)), status_code=303) return RedirectResponse("/software/setup-analyzer/profiles?repository=1&toast_success=" + quote(f"Analyzer profile {profile.get('name', profile_id)} installed."), status_code=303) @app.post("/software/setup-analyzer/analyze") async def setup_analyzer_analyze(request: Request, installer: UploadFile = File(...)): require_admin(request) token, path, size, sha256 = await _save_upload(installer) analysis = analyze_file(path, token, size, sha256) return templates.TemplateResponse("setup_analyzer.html", {"request": request, "analysis": analysis, "max_upload_mb": MAX_UPLOAD_MB, "max_extracted_mb": MAX_EXTRACTED_MB, "max_extracted_files": MAX_EXTRACTED_FILES, "max_sfx_depth": MAX_SFX_DEPTH}) @app.post("/software/setup-analyzer/export/powershell") def setup_analyzer_export_powershell( request: Request, token: str = Form(...), product_name: str = Form(""), product_version: str = Form(""), manufacturer: str = Form(""), architecture: str = Form(""), install_arguments: str = Form(""), timeout_seconds: int = Form(600), run_as: str = Form("system"), success_codes: str = Form("0"), reboot_codes: str = Form(""), suppress_browser: bool = Form(False), process_names: str = Form(""), start_application: bool = Form(False), start_executable: str = Form(""), start_arguments: str = Form(""), start_only_if_user_logged_on: bool = Form(False), start_fail_job_on_error: bool = Form(False), ): require_admin(request) target, meta = _analysis_file(token) values = _export_values(meta, product_name, product_version, manufacturer, architecture, install_arguments, timeout_seconds, run_as, success_codes, reboot_codes, suppress_browser, process_names, start_application, start_executable, start_arguments, start_only_if_user_logged_on, start_fail_job_on_error) deployment_source = _deployment_source_file(target, meta) package = _build_package_manifest(deployment_source, meta, values) package, _profile_notes = normalize_package_manifest(package) script = build_generated_install_script(package) name = _safe_package_name(values["product_name"] or target.stem) headers = {"Content-Disposition": f'attachment; filename="{name}-install.ps1"'} return StreamingResponse(io.BytesIO(script.encode("utf-8")), media_type="text/plain", headers=headers) @app.post("/software/setup-analyzer/export/package") def setup_analyzer_export_package( request: Request, token: str = Form(...), product_name: str = Form(""), product_version: str = Form(""), manufacturer: str = Form(""), architecture: str = Form(""), install_arguments: str = Form(""), timeout_seconds: int = Form(600), run_as: str = Form("system"), success_codes: str = Form("0"), reboot_codes: str = Form(""), suppress_browser: bool = Form(False), process_names: str = Form(""), start_application: bool = Form(False), start_executable: str = Form(""), start_arguments: str = Form(""), start_only_if_user_logged_on: bool = Form(False), start_fail_job_on_error: bool = Form(False), ): require_admin(request) target, meta = _analysis_file(token) values = _export_values(meta, product_name, product_version, manufacturer, architecture, install_arguments, timeout_seconds, run_as, success_codes, reboot_codes, suppress_browser, process_names, start_application, start_executable, start_arguments, start_only_if_user_logged_on, start_fail_job_on_error) deployment_source = _deployment_source_file(target, meta) package = _build_package_manifest(deployment_source, meta, values) package, _profile_notes = normalize_package_manifest(package) install_script = build_generated_install_script(package) detect_script = build_generated_detection_script(package) uninstall_script = build_generated_uninstall_script(package) public_analysis = {key: value for key, value in meta.items() if key != "token"} public_analysis.update(values) stream = io.BytesIO() with zipfile.ZipFile(stream, "w", compression=zipfile.ZIP_DEFLATED) as archive: archive.write(deployment_source, arcname=deployment_source.name) archive.writestr("install.ps1", install_script) archive.writestr("uninstall.ps1", uninstall_script) archive.writestr("detect.ps1", detect_script) archive.writestr("package.json", json.dumps(package, ensure_ascii=True, indent=2)) archive.writestr("analysis.json", json.dumps(public_analysis, ensure_ascii=True, indent=2)) profile_id = str(meta.get("profile_id") or "").strip() if profile_id: try: archive.writestr("metadata/analyzer-profile.amprofile", export_profile_bundle(profile_id)) except Exception: pass stream.seek(0) name = _safe_package_name(values["product_name"] or target.stem) version = _safe_package_name(values["product_version"]) if values["product_version"] else "" filename = f"{name}-{version}.zip" if version else f"{name}.zip" headers = {"Content-Disposition": f'attachment; filename="{filename}"'} return StreamingResponse(stream, media_type="application/zip", headers=headers) @app.post("/software/setup-analyzer/create-package") def setup_analyzer_create_package( request: Request, token: str = Form(...), product_name: str = Form(""), product_version: str = Form(""), manufacturer: str = Form(""), architecture: str = Form(""), install_arguments: str = Form(""), timeout_seconds: int = Form(600), run_as: str = Form("system"), success_codes: str = Form("0"), reboot_codes: str = Form(""), suppress_browser: bool = Form(False), process_names: str = Form(""), start_application: bool = Form(False), start_executable: str = Form(""), start_arguments: str = Form(""), start_only_if_user_logged_on: bool = Form(False), start_fail_job_on_error: bool = Form(False), db: Session = Depends(get_db), ): require_admin(request) target, meta = _analysis_file(token) values = _export_values( meta, product_name, product_version, manufacturer, architecture, install_arguments, timeout_seconds, run_as, success_codes, reboot_codes, suppress_browser, process_names, start_application, start_executable, start_arguments, start_only_if_user_logged_on, start_fail_job_on_error, ) deployment_source = _deployment_source_file(target, meta) manifest = _build_package_manifest(deployment_source, meta, values) manifest, _profile_notes = normalize_package_manifest(manifest) install_script = build_generated_install_script(manifest) detect_script = build_generated_detection_script(manifest) uninstall_script = build_generated_uninstall_script(manifest) public_analysis = {key: value for key, value in meta.items() if key != "token"} public_analysis.update(values) package = SoftwarePackage( name=unique_package_name( db, values["product_name"] or target.stem, values["product_version"], ), description=( f"{values['manufacturer']} | {values['product_name'] or target.stem} " f"{values['product_version']} | Setup Analyzer" ).strip(" |"), package_type="deployment", enabled=True, is_system=False, command_windows="install.ps1", callback_timeout_minutes=max( 5, min(((values["timeout_seconds"] + 59) // 60) + 5, 240), ), ) db.add(package) try: db.flush() manifest["assetmanager_package_id"] = package.id write_package_storage( package.id, deployment_source, install_script, uninstall_script, detect_script, manifest, public_analysis, ) db.commit() except Exception: db.rollback() if package.id: shutil.rmtree(package_directory(package.id), ignore_errors=True) raise return RedirectResponse( f"/software/packages/{package.id}?created=1", status_code=303, )