1782 lines
69 KiB
Python
Executable File
1782 lines
69 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 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<version>\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,
|
|
)
|
|
|