jlengrand/bugsink
1
0
Fork 0
mirror of https://github.com/jlengrand/bugsink.git synced 2026-03-10 08:01:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Switch to safe_join from os.path.join (snappea)

Browse Source 
rather than think-carefully-and-explain-with-nosec, just switch to
safe_join: this saves future readers the pain of validating whether
all assumptions are (still) correct at a (small) performance cost.

See #175
This commit is contained in:
Klaas van Schelven
2025-07-30 14:02:43 +02:00
parent 89db6d2899
commit 17a96a22a6
2 changed files with 6 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
snappea/foreman.py
View File

@@ -14,6 +14,7 @@ import sentry_sdk
from django.conf import settings from django.conf import settings
from django.db import connections from django.db import connections
from django.utils._os import safe_join
from sentry_sdk_extensions import capture_or_log_exception from sentry_sdk_extensions import capture_or_log_exception
from performance.context_managers import time_to_logger from performance.context_managers import time_to_logger
@@ -247,7 +248,7 @@ class Foreman:
# Ensure that anything we might be waiting for is unblocked. A single notification file and .release call is # Ensure that anything we might be waiting for is unblocked. A single notification file and .release call is
# enough because after every wakeup_calls.read() / acquire call in our codebase the first thing we do is # enough because after every wakeup_calls.read() / acquire call in our codebase the first thing we do is
# check_for_stopping(), so the release cannot be inadvertently be "used up" by something else. # check_for_stopping(), so the release cannot be inadvertently be "used up" by something else.
with open(os.path.join(self.settings.WAKEUP_CALLS_DIR, str(uuid.uuid4())), "w"): with open(safe_join(self.settings.WAKEUP_CALLS_DIR, str(uuid.uuid4())), "w"):
pass pass
self.worker_semaphore.release() self.worker_semaphore.release()
@@ -266,7 +267,7 @@ class Foreman:
raise raise
def _run_forever(self): def _run_forever(self):
pre_existing_wakeup_notifications = glob.glob(os.path.join(self.settings.WAKEUP_CALLS_DIR, "*")) pre_existing_wakeup_notifications = glob.glob(safe_join(self.settings.WAKEUP_CALLS_DIR, "*"))
if len(pre_existing_wakeup_notifications) > 0: if len(pre_existing_wakeup_notifications) > 0:
# We clear the wakeup_calls_dir on startup. Not strictly necessary because such files would be cleared by in # We clear the wakeup_calls_dir on startup. Not strictly necessary because such files would be cleared by in
# the loop anyway, but it's more efficient to do it first. # the loop anyway, but it's more efficient to do it first.
@@ -304,7 +305,7 @@ class Foreman:
# such that dogfood Bugsink will pick up on it. But in the end it's not really that relevant (despite # such that dogfood Bugsink will pick up on it. But in the end it's not really that relevant (despite
# being interesting) so I just catch the error. # being interesting) so I just catch the error.
with contextlib.suppress(FileNotFoundError): with contextlib.suppress(FileNotFoundError):
os.unlink(os.path.join(self.settings.WAKEUP_CALLS_DIR, event.name)) os.unlink(safe_join(self.settings.WAKEUP_CALLS_DIR, event.name))
self.check_for_stopping() # check after .read() - it may have unblocked via handle_signal() self.check_for_stopping() # check after .read() - it may have unblocked via handle_signal()
while self.create_workers() == self.settings.TASK_QS_LIMIT: while self.create_workers() == self.settings.TASK_QS_LIMIT:

3
snappea/models.py
View File

@@ -1,6 +1,7 @@
import os import os
from django.db import models from django.db import models
from django.utils._os import safe_join
from .settings import get_settings from .settings import get_settings
from . import thread_uuid from . import thread_uuid
@@ -44,7 +45,7 @@ class Stat(models.Model):
def wakeup_server(): def wakeup_server():
wakeup_file = os.path.join(get_settings().WAKEUP_CALLS_DIR, thread_uuid) wakeup_file = safe_join(get_settings().WAKEUP_CALLS_DIR, thread_uuid)
if not os.path.exists(get_settings().WAKEUP_CALLS_DIR): if not os.path.exists(get_settings().WAKEUP_CALLS_DIR):
os.makedirs(get_settings().WAKEUP_CALLS_DIR, exist_ok=True) os.makedirs(get_settings().WAKEUP_CALLS_DIR, exist_ok=True)