forked from kenzuya/unshackle
fix(monalisa): avoid leaking secrets and add worker safety
- Pass ML-Worker key via env/stdin instead of argv to reduce exposure in process listings/logs. - Add a hard timeout to the ML-Worker subprocess call and convert timeouts into DecryptionFailed errors. - Make ticket bytes decoding defensive: try UTF-8, fall back to ASCII (base64), otherwise raise a descriptive ValueError.
This commit is contained in:
@@ -7,6 +7,7 @@ segment decryption (ML-Worker binary + AES-ECB).
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
@@ -17,6 +18,8 @@ from uuid import UUID
|
|||||||
from Cryptodome.Cipher import AES
|
from Cryptodome.Cipher import AES
|
||||||
from Cryptodome.Util.Padding import unpad
|
from Cryptodome.Util.Padding import unpad
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class MonaLisa:
|
class MonaLisa:
|
||||||
"""
|
"""
|
||||||
@@ -142,7 +145,16 @@ class MonaLisa:
|
|||||||
The raw PSSH value as a base64 string.
|
The raw PSSH value as a base64 string.
|
||||||
"""
|
"""
|
||||||
if isinstance(self._ticket, bytes):
|
if isinstance(self._ticket, bytes):
|
||||||
|
try:
|
||||||
return self._ticket.decode("utf-8")
|
return self._ticket.decode("utf-8")
|
||||||
|
except UnicodeDecodeError:
|
||||||
|
# Tickets are typically base64, so ASCII is a reasonable fallback.
|
||||||
|
try:
|
||||||
|
return self._ticket.decode("ascii")
|
||||||
|
except UnicodeDecodeError as e:
|
||||||
|
raise ValueError(
|
||||||
|
f"Ticket bytes must be UTF-8 text or ASCII base64; got undecodable bytes (len={len(self._ticket)})"
|
||||||
|
) from e
|
||||||
return self._ticket
|
return self._ticket
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@@ -222,19 +234,27 @@ class MonaLisa:
|
|||||||
raise MonaLisa.Exceptions.DecryptionFailed(f"Segment file does not exist: {segment_path}")
|
raise MonaLisa.Exceptions.DecryptionFailed(f"Segment file does not exist: {segment_path}")
|
||||||
|
|
||||||
# Stage 1: ML-Worker decryption
|
# Stage 1: ML-Worker decryption
|
||||||
cmd = [str(worker_path), self._key, str(bbts_path), str(ents_path)]
|
# Do not pass secrets via argv (visible in process listings/logs).
|
||||||
|
# ML-Worker supports receiving the key out-of-band; we provide it via env + stdin.
|
||||||
|
cmd = [str(worker_path), "-", str(bbts_path), str(ents_path)]
|
||||||
|
worker_env = os.environ.copy()
|
||||||
|
worker_env["WORKER_KEY"] = self._key
|
||||||
|
|
||||||
startupinfo = None
|
startupinfo = None
|
||||||
if sys.platform == "win32":
|
if sys.platform == "win32":
|
||||||
startupinfo = subprocess.STARTUPINFO()
|
startupinfo = subprocess.STARTUPINFO()
|
||||||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||||||
|
|
||||||
|
worker_timeout_s = 60
|
||||||
process = subprocess.run(
|
process = subprocess.run(
|
||||||
cmd,
|
cmd,
|
||||||
stdout=subprocess.PIPE,
|
stdout=subprocess.PIPE,
|
||||||
stderr=subprocess.PIPE,
|
stderr=subprocess.PIPE,
|
||||||
text=True,
|
text=True,
|
||||||
|
input=self._key,
|
||||||
|
env=worker_env,
|
||||||
startupinfo=startupinfo,
|
startupinfo=startupinfo,
|
||||||
|
timeout=worker_timeout_s,
|
||||||
)
|
)
|
||||||
|
|
||||||
if process.returncode != 0:
|
if process.returncode != 0:
|
||||||
@@ -260,6 +280,11 @@ class MonaLisa:
|
|||||||
|
|
||||||
except MonaLisa.Exceptions.DecryptionFailed:
|
except MonaLisa.Exceptions.DecryptionFailed:
|
||||||
raise
|
raise
|
||||||
|
except subprocess.TimeoutExpired as e:
|
||||||
|
log.error("ML-Worker timed out after %ss for %s", worker_timeout_s, segment_path.name)
|
||||||
|
raise MonaLisa.Exceptions.DecryptionFailed(
|
||||||
|
f"ML-Worker timed out after {worker_timeout_s}s for {segment_path.name}"
|
||||||
|
) from e
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise MonaLisa.Exceptions.DecryptionFailed(f"Failed to decrypt segment {segment_path.name}: {e}")
|
raise MonaLisa.Exceptions.DecryptionFailed(f"Failed to decrypt segment {segment_path.name}: {e}")
|
||||||
finally:
|
finally:
|
||||||
|
|||||||
Reference in New Issue
Block a user