949 lines
36 KiB
Python
Executable File
949 lines
36 KiB
Python
Executable File
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 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 fastapi import Depends, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.responses import RedirectResponse, StreamingResponse
|
|
|
|
try:
|
|
import pefile
|
|
except Exception:
|
|
pefile = 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")))
|
|
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)
|
|
|
|
|
|
INSTALLER_RULES: list[dict[str, Any]] = [
|
|
{
|
|
"key": "inno",
|
|
"label": "Inno Setup",
|
|
"markers": [
|
|
(b"inno setup setup data", 75, "setup_analyzer.signal.inno_data"),
|
|
(b"inno setup", 35, "setup_analyzer.signal.inno"),
|
|
(b"innosetup", 20, "setup_analyzer.signal.inno_internal"),
|
|
],
|
|
},
|
|
{
|
|
"key": "nsis",
|
|
"label": "NSIS",
|
|
"markers": [
|
|
(b"nullsoft install system", 80, "setup_analyzer.signal.nsis_system"),
|
|
(b"nullsoftinst", 55, "setup_analyzer.signal.nsis_installer"),
|
|
(b"nullsoft", 25, "setup_analyzer.signal.nullsoft"),
|
|
(b"nsis", 20, "setup_analyzer.signal.nsis"),
|
|
],
|
|
},
|
|
{
|
|
"key": "wix_burn",
|
|
"label": "WiX Burn",
|
|
"markers": [
|
|
(b"wixburn", 80, "setup_analyzer.signal.wix_burn"),
|
|
(b"wixbundle", 55, "setup_analyzer.signal.wix_bundle"),
|
|
(b"wixstdba", 45, "setup_analyzer.signal.wix_stdba"),
|
|
(b"burn engine", 30, "setup_analyzer.signal.burn_engine"),
|
|
],
|
|
},
|
|
{
|
|
"key": "installshield",
|
|
"label": "InstallShield",
|
|
"markers": [
|
|
(b"installshield", 80, "setup_analyzer.signal.installshield"),
|
|
(b"installscript", 30, "setup_analyzer.signal.installscript"),
|
|
],
|
|
},
|
|
{
|
|
"key": "advanced_installer",
|
|
"label": "Advanced Installer",
|
|
"markers": [
|
|
(b"advanced installer", 80, "setup_analyzer.signal.advanced_installer"),
|
|
(b"caphyon", 50, "setup_analyzer.signal.caphyon"),
|
|
],
|
|
},
|
|
{
|
|
"key": "squirrel",
|
|
"label": "Squirrel",
|
|
"markers": [
|
|
(b"squirrel", 55, "setup_analyzer.signal.squirrel"),
|
|
(b"releasify", 25, "setup_analyzer.signal.squirrel_releasify"),
|
|
],
|
|
},
|
|
{
|
|
"key": "7zip_sfx",
|
|
"label": "7-Zip SFX",
|
|
"markers": [
|
|
(b"7-zip sfx", 75, "setup_analyzer.signal.7zip_sfx"),
|
|
(b"7zs.sfx", 50, "setup_analyzer.signal.7zip_module"),
|
|
],
|
|
},
|
|
{
|
|
"key": "winrar_sfx",
|
|
"label": "WinRAR SFX",
|
|
"markers": [
|
|
(b"winrar sfx", 75, "setup_analyzer.signal.winrar_sfx"),
|
|
(b"rar sfx", 45, "setup_analyzer.signal.rar_sfx"),
|
|
],
|
|
},
|
|
]
|
|
|
|
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: dict[bytes, str] = {}
|
|
markers = [marker for rule in INSTALLER_RULES for marker, _, _ in rule["markers"]]
|
|
markers.extend(SWITCH_MARKERS)
|
|
for marker in markers:
|
|
lower = marker.lower()
|
|
normalized = lower.decode("ascii", errors="ignore")
|
|
result[lower] = normalized
|
|
result[normalized.encode("utf-16le")] = normalized
|
|
return result
|
|
|
|
|
|
SCAN_NEEDLES = _scan_needles()
|
|
|
|
|
|
def _scan_file_markers(path: Path) -> set[str]:
|
|
found: set[str] = set()
|
|
tail = b""
|
|
with path.open("rb") as handle:
|
|
while True:
|
|
chunk = handle.read(SCAN_CHUNK_BYTES)
|
|
if not chunk:
|
|
break
|
|
data = (tail + chunk).lower()
|
|
for raw, normalized in SCAN_NEEDLES.items():
|
|
if normalized not in found and raw in data:
|
|
found.add(normalized)
|
|
tail = data[-SCAN_OVERLAP_BYTES:]
|
|
return found
|
|
|
|
|
|
def _pe_metadata(path: Path) -> dict[str, Any]:
|
|
result: dict[str, Any] = {
|
|
"architecture": "",
|
|
"company_name": "",
|
|
"product_name": "",
|
|
"product_version": "",
|
|
"file_version": "",
|
|
"original_filename": "",
|
|
"signature_present": None,
|
|
}
|
|
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: list[dict[str, Any]] = []
|
|
for rule in INSTALLER_RULES:
|
|
score = 0
|
|
signals: list[str] = []
|
|
for marker, points, signal_key in rule["markers"]:
|
|
if marker.decode("ascii").lower() in found_markers:
|
|
score += points
|
|
signals.append(signal_key)
|
|
if score:
|
|
candidates.append({
|
|
"key": rule["key"],
|
|
"label": rule["label"],
|
|
"confidence": min(score, 99),
|
|
"signals": signals,
|
|
})
|
|
candidates.sort(key=lambda item: item["confidence"], reverse=True)
|
|
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["signals"]),
|
|
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,
|
|
"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 installer_type == "inno":
|
|
args = "/VERYSILENT /SUPPRESSMSGBOXES /NORESTART /SP-"
|
|
result.update(install_arguments=args, install_command=f"{quoted} {args}", success_codes=[0, 3010], reboot_codes=[3010], command_confidence="high")
|
|
elif installer_type == "nsis":
|
|
args = "/S"
|
|
result.update(install_arguments=args, install_command=f"{quoted} {args}", success_codes=[0, 3010], reboot_codes=[3010], command_confidence="high")
|
|
elif installer_type == "wix_burn":
|
|
args = "/quiet /norestart"
|
|
result.update(install_arguments=args, install_command=f"{quoted} {args}", success_codes=[0, 1641, 3010], reboot_codes=[1641, 3010], command_confidence="high")
|
|
elif installer_type == "installshield":
|
|
args = '/s /v"/qn /norestart"'
|
|
result.update(install_arguments=args, install_command=f"{quoted} {args}", success_codes=[0, 1641, 3010], reboot_codes=[1641, 3010], command_confidence="medium")
|
|
result["warning_keys"].append("setup_analyzer.warning.installshield")
|
|
elif installer_type == "advanced_installer":
|
|
args = "/exenoui /qn /norestart"
|
|
result.update(install_arguments=args, install_command=f"{quoted} {args}", success_codes=[0, 1641, 3010], reboot_codes=[1641, 3010], command_confidence="medium")
|
|
result["warning_keys"].append("setup_analyzer.warning.advanced_installer")
|
|
elif installer_type == "squirrel":
|
|
args = "--silent"
|
|
result.update(install_arguments=args, install_command=f"{quoted} {args}", success_codes=[0], command_confidence="low")
|
|
result["warning_keys"].append("setup_analyzer.warning.squirrel")
|
|
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 {"7zip_sfx", "winrar_sfx"}:
|
|
result["warning_keys"].append("setup_analyzer.warning.sfx")
|
|
else:
|
|
result["warning_keys"].append("setup_analyzer.warning.unknown")
|
|
return result
|
|
|
|
|
|
def analyze_file(path: Path, token: str, size: int, sha256: str) -> 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]] = []
|
|
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)
|
|
|
|
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 ""
|
|
manufacturer = msi.get("Manufacturer") or msix.get("publisher") or pe.get("company_name") or ""
|
|
architecture = msix.get("architecture") or pe.get("architecture") or ""
|
|
product_code = msi.get("ProductCode", "")
|
|
upgrade_code = msi.get("UpgradeCode", "")
|
|
identity_text = f"{filename} {product_name} {manufacturer}".casefold()
|
|
if "greenshot" in identity_text:
|
|
product_name = product_name or "Greenshot"
|
|
manufacturer = manufacturer or "Greenshot"
|
|
if not product_version:
|
|
version_match = re.search(r"(?i)greenshot[-_ ]installer[-_ ](\d+(?:\.\d+){1,3})", filename)
|
|
if version_match:
|
|
product_version = version_match.group(1)
|
|
defaults = _command_defaults(installer_type, filename, product_code, product_name)
|
|
if installer_type == "inno" and "greenshot" in identity_text:
|
|
arguments = str(defaults.get("install_arguments") or "").strip()
|
|
if not re.search(r"(?i)(^|\s)/(ALLUSERS|CURRENTUSER)(?=\s|$)", arguments):
|
|
arguments += " /ALLUSERS"
|
|
if not re.search(r"(?i)(^|\s)/DIR=", arguments):
|
|
arguments += ' /DIR="C:\\Program Files\\Greenshot"'
|
|
defaults["install_arguments"] = arguments.strip()
|
|
defaults["install_command"] = f'"{filename}" {arguments.strip()}'
|
|
defaults["detect_method"] = "registry_display_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")
|
|
|
|
embedded_switches = [marker.decode("ascii") for marker in SWITCH_MARKERS if marker.decode("ascii").lower() in found_markers]
|
|
identity_text = f"{filename} {product_name} {manufacturer}".casefold()
|
|
suppress_browser_default = "greenshot" in identity_text
|
|
process_names_default = "Greenshot.exe" if "greenshot" in identity_text else ""
|
|
analysis = {
|
|
"token": token,
|
|
"filename": filename,
|
|
"size": size,
|
|
"size_human": _human_size(size),
|
|
"sha256": sha256,
|
|
"extension": extension,
|
|
"installer_type": installer_type,
|
|
"installer_label": installer_label,
|
|
"confidence": confidence,
|
|
"candidates": candidates,
|
|
"product_name": product_name,
|
|
"product_version": product_version,
|
|
"manufacturer": manufacturer,
|
|
"architecture": architecture,
|
|
"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": suppress_browser_default,
|
|
"process_names_default": process_names_default,
|
|
"warning_keys": warning_keys,
|
|
"analyzed_at": datetime.now(timezone.utc).isoformat(),
|
|
"msiinfo_available": shutil.which("msiinfo") is not None,
|
|
"pefile_available": pefile is not None,
|
|
**defaults,
|
|
}
|
|
(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 = "",
|
|
) -> 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),
|
|
}
|
|
|
|
|
|
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",
|
|
"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,
|
|
},
|
|
"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"),
|
|
},
|
|
}
|
|
|
|
|
|
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})
|
|
|
|
@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})
|
|
|
|
@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(""),
|
|
):
|
|
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)
|
|
package = _build_package_manifest(target, 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(""),
|
|
):
|
|
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)
|
|
package = _build_package_manifest(target, 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(target, arcname=target.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))
|
|
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(""),
|
|
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,
|
|
)
|
|
manifest = _build_package_manifest(target, 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,
|
|
target,
|
|
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,
|
|
)
|
|
|