3 Commits

Author SHA1 Message Date
Naiel
80e9262bcb fix logic 2026-03-03 11:33:23 +01:00
Naiel
cd456ab9f1 fix 2026-03-03 11:29:38 +01:00
Naiel
c0c40ecd99 fix 2026-03-03 11:24:19 +01:00

View File

@@ -84,32 +84,52 @@ def ts_encrypt(input_value: Any, secret: str) -> str:
return f"RSA{{{b64}}}"
def ts_encrypt(input_value: Any, secret: str) -> str:
def ts_decrypt(input_value: Any, secret: str) -> Any:
"""
Compatible with JS TS_decrypt behavior:
- If not string: return as-is.
- If RSA{...}: decrypt AES(CryptoJS passphrase mode), parse JSON when possible.
- If plain string JSON: parse JSON.
- Else: return raw string.
"""
if not isinstance(input_value, str):
payload = json.dumps(input_value, separators=(",", ":"), ensure_ascii=False)
else:
payload = input_value
return input_value
payload_bytes = payload.encode("utf-8")
salt = os.urandom(8)
is_wrapped = input_value.startswith("RSA{") and input_value.endswith("}")
if is_wrapped:
if not secret:
raise TeleSecCryptoError("Secret is required to decrypt RSA payload")
b64 = input_value[4:-1]
try:
raw = base64.b64decode(b64)
except Exception as exc:
raise TeleSecCryptoError("Invalid base64 payload") from exc
# OpenSSL EVP_BytesToKey (MD5)
dx = b""
salted = b""
while len(salted) < 48: # 32 key + 16 iv
dx = hashlib.md5(dx + secret.encode() + salt).digest()
salted += dx
if len(raw) < 16 or not raw.startswith(b"Salted__"):
raise TeleSecCryptoError("Unsupported encrypted payload format")
key = salted[:32]
iv = salted[32:48]
salt = raw[8:16]
ciphertext = raw[16:]
key, iv = _evp_bytes_to_key(secret.encode("utf-8"), salt, 32, 16)
cipher = AES.new(key, AES.MODE_CBC, iv=iv)
decrypted = cipher.decrypt(ciphertext)
decrypted = _pkcs7_unpad(decrypted, 16)
cipher = AES.new(key, AES.MODE_CBC, iv)
encrypted = cipher.encrypt(_pkcs7_pad(payload_bytes, 16))
try:
text = decrypted.decode("utf-8")
except UnicodeDecodeError:
text = decrypted.decode("latin-1")
openssl_blob = b"Salted__" + salt + encrypted
b64 = base64.b64encode(openssl_blob).decode("utf-8")
try:
return json.loads(text)
except Exception:
return text
return f"RSA{{{b64}}}"
try:
return json.loads(input_value)
except Exception:
return input_value
@dataclass
class TeleSecDoc:
@@ -352,7 +372,7 @@ def should_shutdown(data: Dict[str, Any], server_now: datetime) -> bool:
target = parse_iso(str(data.get("ShutdownBeforeDate", "") or ""))
if not target:
return False
return server_now >= target
return server_now <= target
def execute_shutdown(dry_run: bool = False) -> None:
@@ -401,8 +421,8 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--secret", default="", help="TeleSec secret para cifrado")
parser.add_argument("--machine-id", default="", help="ID de máquina (default: hostname)")
parser.add_argument("--interval", type=int, default=15, help="Intervalo en segundos")
parser.add_argument("--once", action="store_false", help="Ejecutar una sola iteración")
parser.add_argument("--dry-run", action="store_false", help="No apagar realmente, solo log")
parser.add_argument("--once", action="store_true", help="Ejecutar una sola iteración")
parser.add_argument("--dry-run", action="store_true", help="No apagar realmente, solo log")
parser.add_argument(
"--config",
default="",