fix(OMNY): use requests instead of shelling out to wget for tomo ID
CI for csaxs_bec / test (push) Failing after 11m43s

TomoIDManager.register() called wget via subprocess with shell=True,
interpolating sample_name/eaccount/etc. unescaped into the command
string -- broke outright wherever wget isn't installed (as hit while
testing: "wget: command not found", silently falling back to tomo ID
0), and was a latent shell-injection risk. Use requests.get() with a
params dict instead: no external binary dependency, proper URL
encoding, and the same self-signed-cert/SSRF handling already used by
the samples PDF upload. Drops the now-unused OMNYToolsError class and
TMP_FILE/subprocess, which only existed to support the wget call.
This commit is contained in:
x01dc
2026-07-27 17:06:10 +02:00
parent 3422eedcc1
commit eb61df1c93
@@ -4,7 +4,6 @@ import fcntl
import json
import os
import socket
import subprocess
import sys
import termios
import threading
@@ -36,10 +35,6 @@ def umvr(*args):
return scans.umv(*args, relative=True)
class OMNYToolsError(Exception):
pass
class OMNYTools:
HEADER = "\033[95m"
@@ -354,7 +349,6 @@ class TomoIDManager:
OMNY_URL = "https://v1p0zyg2w9n2k9c1.myfritz.net/samples/newmeasurement.php"
TEST_OMNY_URL = "https://omny-test.psi.ch/samples/newmeasurement.php"
TMP_FILE = "~/currsamplesnr.txt"
FALLBACK_TOMO_ID = 0
@staticmethod
@@ -389,28 +383,39 @@ class TomoIDManager:
"production -- the eaccount recorded there won't be real."
)
url = (
f"{omny_url}"
f"?sample={sample_name}"
f"&date={date}"
f"&eaccount={eaccount}"
f"&scannr={scan_number}"
f"&setup={setup}"
f"&additional={additional_info}"
f"&user={user}"
)
params = {
"sample": sample_name,
"date": date,
"eaccount": eaccount,
"scannr": scan_number,
"setup": setup,
"additional": additional_info,
"user": user,
}
tmp_file = os.path.expanduser(self.TMP_FILE)
try:
result = subprocess.run(f"wget -q -O {tmp_file} '{url}'", shell=True, timeout=30)
if result.returncode != 0:
raise OMNYToolsError(
f"wget failed (exit code {result.returncode}) fetching tomo ID from {omny_url}"
)
with open(tmp_file) as f:
content = f.read().strip()
return int(content)
except (subprocess.TimeoutExpired, FileNotFoundError, ValueError, OMNYToolsError) as exc:
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
except ImportError as exc:
logger.warning(
f"Could not obtain tomo ID from OMNY database ('requests' library not "
f"installed: {exc}); falling back to tomo ID {self.FALLBACK_TOMO_ID}."
)
return self.FALLBACK_TOMO_ID
try:
response = requests.get(
omny_url,
params=params,
timeout=30,
verify=False, # accept self-signed certs
allow_redirects=False, # SSRF hardening
)
response.raise_for_status()
return int(response.text.strip())
except (requests.RequestException, ValueError) as exc:
logger.warning(
f"Could not obtain tomo ID from OMNY database ({exc}); "
f"falling back to tomo ID {self.FALLBACK_TOMO_ID}."