ruff: Fix S108 Probable insecure usage of temporary file.

Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg 2023-01-06 00:09:53 -08:00 committed by Tim Abbott
parent d05f672132
commit 7e3a681f80
6 changed files with 16 additions and 11 deletions

View File

@ -3,6 +3,7 @@ import os
import signal import signal
import subprocess import subprocess
import sys import sys
import tempfile
import time import time
from typing import Tuple from typing import Tuple
@ -37,12 +38,11 @@ def start_server(logfile_name: str) -> Tuple[bool, str]:
if __name__ == "__main__": if __name__ == "__main__":
print("Testing development server start!") print("Testing development server start!")
logfile_name = "/tmp/run-dev-output" with tempfile.NamedTemporaryFile(buffering=0) as logfile:
with open(logfile_name, "wb", buffering=0) as logfile:
run_dev = subprocess.Popen( run_dev = subprocess.Popen(
[os.path.join(TOOLS_DIR, "run-dev.py")], stdout=logfile, stderr=subprocess.STDOUT [os.path.join(TOOLS_DIR, "run-dev.py")], stdout=logfile, stderr=subprocess.STDOUT
) )
failure, log = start_server(logfile_name) failure, log = start_server(logfile.name)
run_dev.send_signal(signal.SIGINT) run_dev.send_signal(signal.SIGINT)
run_dev.wait() run_dev.wait()

View File

@ -85,7 +85,8 @@ def tracemalloc_listen() -> None:
listener_pid = os.getpid() listener_pid = os.getpid()
sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
path = f"/tmp/tracemalloc.{os.getpid()}" os.makedirs(settings.TRACEMALLOC_DUMP_DIR, exist_ok=True)
path = os.path.join(settings.TRACEMALLOC_DUMP_DIR, f"tracemalloc.{os.getpid()}")
sock.bind(path) sock.bind(path)
thread = threading.Thread(target=lambda: tracemalloc_listen_sock(sock), daemon=True) thread = threading.Thread(target=lambda: tracemalloc_listen_sock(sock), daemon=True)
thread.start() thread.start()
@ -96,7 +97,7 @@ def maybe_tracemalloc_listen() -> None:
"""If tracemalloc tracing enabled, listen for requests to dump a snapshot. """If tracemalloc tracing enabled, listen for requests to dump a snapshot.
To trigger once this is listening: To trigger once this is listening:
echo | socat -u stdin unix-sendto:/tmp/tracemalloc.$pid echo | socat -u stdin unix-sendto:/var/log/zulip/tracemalloc/tracemalloc.$pid
To enable in the Zulip web server: edit /etc/zulip/uwsgi.ini , To enable in the Zulip web server: edit /etc/zulip/uwsgi.ini ,
and add e.g. ` PYTHONTRACEMALLOC=5` to the `env=` line. and add e.g. ` PYTHONTRACEMALLOC=5` to the `env=` line.

View File

@ -1,5 +1,6 @@
import cProfile import cProfile
import logging import logging
import tempfile
import time import time
import traceback import traceback
from typing import Any, AnyStr, Callable, Dict, Iterable, List, MutableMapping, Optional, Tuple from typing import Any, AnyStr, Callable, Dict, Iterable, List, MutableMapping, Optional, Tuple
@ -277,8 +278,11 @@ def write_log_line(
if settings.PROFILE_ALL_REQUESTS: if settings.PROFILE_ALL_REQUESTS:
log_data["prof"].disable() log_data["prof"].disable()
profile_path = "/tmp/profile.data.{}.{}".format(path.split("/")[-1], int(time_delta * 1000)) with tempfile.NamedTemporaryFile(
log_data["prof"].dump_stats(profile_path) prefix="profile.data.{}.{}.".format(path.split("/")[-1], int(time_delta * 1000)),
delete=False,
) as stats_file:
log_data["prof"].dump_stats(stats_file.name)
# Log some additional data whenever we return certain 40x errors # Log some additional data whenever we return certain 40x errors
if 400 <= status_code < 500 and status_code not in [401, 404, 405]: if 400 <= status_code < 500 and status_code not in [401, 404, 405]:

View File

@ -55,7 +55,7 @@ class RealmExportTest(ZulipTestCase):
args = mock_export.call_args_list[0][1] args = mock_export.call_args_list[0][1]
self.assertEqual(args["realm"], admin.realm) self.assertEqual(args["realm"], admin.realm)
self.assertEqual(args["public_only"], True) self.assertEqual(args["public_only"], True)
self.assertIn("/tmp/zulip-export-", args["output_dir"]) self.assertTrue(os.path.basename(args["output_dir"]).startswith("zulip-export-"))
self.assertEqual(args["threads"], 6) self.assertEqual(args["threads"], 6)
# Get the entry and test that iago initiated it. # Get the entry and test that iago initiated it.
@ -125,7 +125,7 @@ class RealmExportTest(ZulipTestCase):
args = mock_export.call_args_list[0][1] args = mock_export.call_args_list[0][1]
self.assertEqual(args["realm"], admin.realm) self.assertEqual(args["realm"], admin.realm)
self.assertEqual(args["public_only"], True) self.assertEqual(args["public_only"], True)
self.assertIn("/tmp/zulip-export-", args["output_dir"]) self.assertTrue(os.path.basename(args["output_dir"]).startswith("zulip-export-"))
self.assertEqual(args["threads"], 6) self.assertEqual(args["threads"], 6)
# Get the entry and test that iago initiated it. # Get the entry and test that iago initiated it.

View File

@ -1080,7 +1080,7 @@ class TestWorker(QueueProcessingWorker):
# This worker allows you to test the queue worker infrastructure without # This worker allows you to test the queue worker infrastructure without
# creating significant side effects. It can be useful in development or # creating significant side effects. It can be useful in development or
# for troubleshooting prod/staging. It pulls a message off the test queue # for troubleshooting prod/staging. It pulls a message off the test queue
# and appends it to a file in /tmp. # and appends it to a file in /var/log/zulip.
def consume(self, event: Mapping[str, Any]) -> None: # nocoverage def consume(self, event: Mapping[str, Any]) -> None: # nocoverage
fn = settings.ZULIP_WORKER_TEST_FILE fn = settings.ZULIP_WORKER_TEST_FILE
message = orjson.dumps(event) message = orjson.dumps(event)

View File

@ -694,7 +694,7 @@ RETENTION_LOG_PATH = zulip_path("/var/log/zulip/message_retention.log")
AUTH_LOG_PATH = zulip_path("/var/log/zulip/auth.log") AUTH_LOG_PATH = zulip_path("/var/log/zulip/auth.log")
SCIM_LOG_PATH = zulip_path("/var/log/zulip/scim.log") SCIM_LOG_PATH = zulip_path("/var/log/zulip/scim.log")
ZULIP_WORKER_TEST_FILE = "/tmp/zulip-worker-test-file" ZULIP_WORKER_TEST_FILE = zulip_path("/var/log/zulip/zulip-worker-test-file")
if IS_WORKER: if IS_WORKER: