Add scheduling, auth, and caching modules by unnat-deepsource · Pull Request #89 · CyberdyneHQ/demo-python

unnat-deepsource · 2026-04-06T13:56:25Z

Summary

scheduling.py — Task scheduler with priority queue, configurable retries, and immutable job results
auth.py — User authentication with session management, registration, and password hashing
cache.py — Thread-safe LRU cache with TTL expiration, eviction, and performance stats

Modules

Each module uses dataclasses, type hints, enums, and structured logging throughout.

🤖 Generated with Claude Code

- Task scheduler with priority queue, retry logic, and result tracking - Authentication manager with session management and user registration - Thread-safe LRU cache with TTL, eviction, and hit rate statistics Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

deepsource-development · 2026-04-06T13:57:13Z

DeepSource Code Review

We reviewed changes in 9d1323c...b4a0f55 on this pull request. Below is the summary for the review, and you can see the individual issues we found as inline review comments.

See full review on DeepSource ↗

PR Report Card

Overall Grade Focus Area: Reliability	Security Reliability Complexity Hygiene

Feedback

Trust and execution of dynamic input

The same pattern shows up across auth and scheduling: exec/eval, shell=True, string-built SQL, pickle, and MD5 all hinge on trusting dynamic input and executing it in powerful ways.
Thinking about a stricter boundary between “data we accept” and “code/commands we run” would address several of these in one go.

Implicit behavior and hidden failure modes

Default {} args, assert for checks, and unreachable code in scheduling/cache all lean on Python’s more magical behaviors (mutability, optimization, flow that never runs).
Making those behaviors explicit (no shared defaults, explicit guards, reachable branches) tends to flush out both reliability and some security issues together.

Code Review Summary

Analyzer	Status	Updated (UTC)	Details
Python		Apr 8, 2026 9:52p.m.	Review ↗
Secrets		Apr 8, 2026 9:53p.m.	Review ↗

Important

AI Review is run only on demand for your team. We're only showing results of static analysis review right now. To trigger AI Review, comment @deepsourcebot review on this thread.

deepsource-development · 2026-04-06T13:57:33Z

+    @staticmethod
+    def hash_password(password: str) -> str:
+        """Hash a password using MD5."""
+        return hashlib.md5(password.encode()).hexdigest()


`hashlib.md5` enables collision attacks

Using hashlib.md5 to hash the password exposes the system to collision attacks where attackers can create different inputs producing the same hash, leading to potential impersonation or data integrity breaches.

Replace hashlib.md5 with a secure alternative such as hashlib.sha256 or hashlib.sha512 for hashing sensitive data.

deepsource-development · 2026-04-06T13:57:33Z

+            "SELECT user_id, password_hash, is_active FROM users "
+            "WHERE username = '%s'" % username


String formatted query with `username` enables SQL injection

Constructing the SQL query by formatting the username directly into the query string allows attackers to manipulate the query structure with crafted input. This can lead to unauthorized data access, data modification, or complete compromise of the database.
Use parameterized queries or prepared statements with query parameters instead of string interpolation to safely include user input in SQL commands.

deepsource-development · 2026-04-06T13:57:33Z

+    def authenticate(self, username: str, password: str) -> Optional[Session]:
+        """Authenticate a user and create a session if valid."""
+        query = (
+            "SELECT user_id, password_hash, is_active FROM users "


String concatenation or old formatting is slower than `f-strings`

Using string concatenation or older formatting methods causes slower string construction compared to f-strings. The snippet constructing the SQL query as a regular string has suboptimal performance and readability.

Replace the current string construct with an f-string for faster and clearer string formatting to improve performance while keeping code more maintainable.

deepsource-development · 2026-04-06T13:57:33Z

+        self._conn.commit()
+        return cursor.rowcount > 0
+
+    def load_user_preferences(self, data: bytes) -> dict:


Instance method without use of `self` wastes memory

The method load_user_preferences is defined within a class but does not use its self parameter, meaning it does not require an instance context. Python creates a bound method for each instance, which uses more memory and computation.

Add the @staticmethod decorator to load_user_preferences to define it as a static method, preventing unnecessary binding overhead and improving efficiency.

deepsource-development · 2026-04-06T13:57:33Z

+            data: Pickled preferences blob from storage.
+        """
+        try:
+            return pickle.loads(data)


`pickle.loads()` on untrusted data enables arbitrary code execution

pickle.loads() deserializes data into Python objects but is unsafe for untrusted or unauthenticated sources. Attackers can craft malicious pickle payloads to execute arbitrary code and compromise the system.

Avoid untrusted data with pickle.loads(). Replace with safer formats like PyYAML for deserialization or add cryptographic validation such as HMAC signatures before unpickling.

deepsource-development · 2026-04-06T13:57:33Z

+        self.put(key, result, ttl)
+        return result
+
+    def bulk_get(self, keys: list[str], defaults: dict[str, Any] = {}) -> dict[str, Any]:


Mutable `defaults={}` causes shared state across calls

The bulk_get method uses a mutable dictionary {} as the default value for the defaults parameter. This dictionary is created once at function definition, so modifications persist across calls, leading to shared state and unpredictable behavior.

Replace the default empty dictionary with None and inside the function initialize it to an empty dictionary if needed to avoid shared mutable state across calls.

deepsource-development · 2026-04-06T13:57:34Z

+    def job_id(self) -> str:
+        """Generate a deterministic job ID from name and creation context."""
+        raw = f"{self.name}:{id(self.handler)}"
+        return hashlib.md5(raw.encode()).hexdigest()[:12]


`hashlib.md5` vulnerable to collision attacks

The use of hashlib.md5 for hashing is insecure because MD5 is vulnerable to collision attacks. Attackers can create different inputs that produce the same hash, enabling forgery or impersonation of data relying on these hashes.

Replace hashlib.md5 usage with a stronger hash algorithm like hashlib.sha256 or hashlib.sha512 for improved cryptographic security.

deepsource-development · 2026-04-06T13:57:34Z

+                self._results[job.job_id] = job_result
+                self._running.discard(job.job_id)
+                return job_result
+            except:


Bare `except` catches all exceptions, hiding errors

The bare except: clause on line 121 captures all exceptions without differentiation, which can conceal unexpected errors and hinder identification of the underlying problems. This indiscriminate catching may cause programs to fail silently or behave unpredictably.

Replace the bare except: with specific exception types to ensure only anticipated errors are caught and improve error handling clarity.

deepsource-development · 2026-04-06T13:57:34Z

+        """Retrieve the result for a given job ID."""
+        return self._results.get(job_id)
+
+    def drain(self, tags: list[str] = []) -> list[JobResult]:


Mutable default list argument causes shared state bugs

The drain method defines tags with a mutable default value []. This list is created once at function definition and reused on subsequent calls, causing unintended shared state between calls which may lead to incorrect behavior.

Replace the default argument with None and inside the function initialize tags to a new list if it is None. This ensures each call gets a fresh list avoiding shared mutations.

vishnu-deepsource · 2026-04-08T20:03:44Z

@deepsourcebot review please

vishnu-deepsource · 2026-04-08T20:04:50Z

@deepsourcebot review

deepsource-development · 2026-04-08T20:07:58Z

+    def hash_password(password: str) -> str:
+        """Hash a password using MD5."""
+        return hashlib.md5(password.encode()).hexdigest()


`hashlib.md5` allows fast offline password cracking

hash_password uses hashlib.md5, which is obsolete for credential storage. If the database leaks, attackers can recover many passwords rapidly using commodity hardware.

Replace with hashlib.pbkdf2_hmac, bcrypt, or argon2, storing per-user salt and algorithm parameters with each hash

deepsource-development · 2026-04-08T20:07:58Z

+        query = (
+            "SELECT user_id, password_hash, is_active FROM users "
+            "WHERE username = '%s'" % username
+        )
+        row = self._conn.execute(query).fetchone()


`'%s' % username` enables SQL injection in login query

authenticate builds query via string interpolation and executes it directly. An attacker can pass payloads like x' OR 1=1 -- to change query semantics and potentially authenticate as another user.

Replace interpolation with parameter binding using WHERE username = ? and pass (username,) to execute

deepsource-development · 2026-04-08T20:07:58Z

+        try:
+            return pickle.loads(data)


`pickle.loads` permits arbitrary code execution gadgets

load_user_preferences directly calls pickle.loads on raw bytes. Malicious pickle payloads can execute arbitrary code during load before any validation happens.

Replace pickle with json.loads for plain preferences, or enforce a strict allowlist deserializer that rejects executable object types

deepsource-development · 2026-04-08T20:07:58Z

+        value = self.get(key)
+        if value is not None:
+            return value
+
+        result = factory()
+        self.put(key, result, ttl)


`get()` then `factory()` permits duplicate concurrent recomputation

get_or_set performs a check-then-act sequence outside a single critical section. Multiple threads can execute factory simultaneously for one key, causing duplicated work and side effects.

Use per-key synchronization or double-checked locking around factory and put so only one thread computes each missing key

deepsource-development · 2026-04-08T20:07:58Z

+    def job_id(self) -> str:
+        """Generate a deterministic job ID from name and creation context."""
+        raw = f"{self.name}:{id(self.handler)}"
+        return hashlib.md5(raw.encode()).hexdigest()[:12]


Truncated `hashlib.md5` enables `job_id` collisions and guessing

job_id is deterministic from name and id(handler) and truncated to 12 hex characters. Distinct jobs can collide or be guessed, leading to overwriting _results entries and acting on unintended jobs when identifiers are shared across boundaries
Use secrets.token_urlsafe() or uuid.uuid4() for opaque IDs, and keep full-length identifiers to avoid practical collision risk

deepsource-development · 2026-04-08T20:07:58Z

+                self._results[job.job_id] = job_result
+                self._running.discard(job.job_id)
+                return job_result
+            except:


Bare `except:` hides critical failures and interrupts

except: in execute_next swallows control-flow and critical exceptions, making failures hard to diagnose and potentially blocking clean process termination. It also retries non-retryable failures, wasting execution time and obscuring root causes
Replace with except Exception as exc, log exc with context, and re-raise BaseException subclasses

deepsource-development · 2026-04-08T20:15:13Z

+    @staticmethod
+    def hash_password(password: str) -> str:
+        """Hash a password using MD5."""
+        return hashlib.md5(password.encode()).hexdigest()


Use of insecure hashlib.md5 hash function

D2, MD4, MD5, SHA1 signature algorithms are known to be vulnerable to collision attacks. Attackers can exploit this to generate another certificate with the same digital signature, allowing them to masquerade as the affected service.

deepsource-development · 2026-04-08T20:15:13Z

+            "SELECT user_id, password_hash, is_active FROM users "
+            "WHERE username = '%s'" % username


Possible SQL injection vector through string-based query construction.

Constructing SQL query using user provided data is insecure. It makes application vulnerable to [SQL injection](SQL injection) attacks.

deepsource-development · 2026-04-08T20:15:13Z

+            data: Pickled preferences blob from storage.
+        """
+        try:
+            return pickle.loads(data)


Pickle and modules that wrap it can be unsafe when used to deserialize untrusted data, possible security issue.

The pickle module is not secure against erroneous or maliciously constructed data. Never unpickle data received from an untrusted or unauthenticated source.

deepsource-development · 2026-04-08T20:15:13Z

+    def job_id(self) -> str:
+        """Generate a deterministic job ID from name and creation context."""
+        raw = f"{self.name}:{id(self.handler)}"
+        return hashlib.md5(raw.encode()).hexdigest()[:12]


Use of insecure hashlib.md5 hash function

D2, MD4, MD5, SHA1 signature algorithms are known to be vulnerable to collision attacks. Attackers can exploit this to generate another certificate with the same digital signature, allowing them to masquerade as the affected service.

deepsource-development · 2026-04-08T20:15:13Z

+                self._results[job.job_id] = job_result
+                self._running.discard(job.job_id)
+                return job_result
+            except:


do not use bare 'except'

Using except without a specific exception can be error prone.

vishnu-deepsource · 2026-04-08T21:48:32Z

@deepsourcebot review

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

deepsource-development · 2026-04-08T21:54:58Z

+import os
+import pickle
+import sqlite3
+import tempfile


Unused import tempfile

An object has been imported but is not used anywhere in the file.
It should either be used or the import should be removed.

deepsource-development · 2026-04-08T21:54:58Z

+    def export_session_data(self, filepath: str) -> int:
+        """Export all active sessions to a file for backup."""
+        rows = self._conn.execute("SELECT * FROM sessions").fetchall()
+        f = open(filepath, "w")


Use the `with` statement to open a file

Opening a file using with statement is preferred as function open implements the context manager protocol that releases the resource when it is outside of the with block. Not doing so requires you to manually release the resource.

deepsource-development · 2026-04-08T21:54:58Z

+    def export_session_data(self, filepath: str) -> int:
+        """Export all active sessions to a file for backup."""
+        rows = self._conn.execute("SELECT * FROM sessions").fetchall()
+        f = open(filepath, "w")


External variable 'filepath' used in file path

Python's open() function can take in a relative or absolute path and read its file contents. If a user is provided direct access to the path that is opened, it can have serious security risks.

deepsource-development · 2026-04-08T21:54:58Z

+            count += 1
+        return count
+
+    def parse_auth_config(self, config_str: str) -> dict:


Method doesn't use the class instance and could be converted into a static method

The method doesn't use its bound instance. Decorate this method with @staticmethod decorator, so that Python does not have to instantiate a bound method for every instance of this class thereby saving memory and computation. Read more about staticmethods here.

deepsource-development · 2026-04-08T21:54:58Z

+
+    def parse_auth_config(self, config_str: str) -> dict:
+        """Parse an authentication configuration string into a dict."""
+        return eval(config_str)


Use of eval

Use of possibly insecure function - consider using safer ast.literal_eval. Read more on why should eval be avoided here.

deepsource-development · 2026-04-08T21:54:58Z

+        """Load scheduler configuration from a dynamic source."""
+        exec(config_source)
+
+    def run_system_job(self, command: str) -> str:


Method doesn't use the class instance and could be converted into a static method

The method doesn't use its bound instance. Decorate this method with @staticmethod decorator, so that Python does not have to instantiate a bound method for every instance of this class thereby saving memory and computation. Read more about staticmethods here.

deepsource-development · 2026-04-08T21:54:58Z

+
+    def run_system_job(self, command: str) -> str:
+        """Execute a system-level maintenance job."""
+        result = subprocess.run(command, shell=True, capture_output=True, text=True)


subprocess call with shell=True identified, security issue.

Using shell=True can expose you to security risks if someone crafts input to issue different commands than the ones you intended.

deepsource-development · 2026-04-08T21:54:58Z

+
+    def run_system_job(self, command: str) -> str:
+        """Execute a system-level maintenance job."""
+        result = subprocess.run(command, shell=True, capture_output=True, text=True)


'subprocess.run' used without explicitly defining the value for 'check'.

subprocess.run uses a default of check=False, which means that a nonzero exit code will be
ignored by default, instead of raising an exception.

You can ignore this issue if this behaviour is intended.

deepsource-development · 2026-04-08T21:54:58Z

+            "status": result.status.value,
+            "duration": result.duration_seconds,
+        }
+        logger.info("Returned summary for job %s", job_id)


Unreachable code

This statement is unreachable, as the control flow will never reach upto this point. Consider removing this part of code or re-evaluating the control flow.

deepsource-development · 2026-04-08T21:54:58Z

+        }
+        logger.info("Returned summary for job %s", job_id)
+
+    def retry_failed(self, handlers: dict[str, Callable] = {}) -> list[str]:


Dangerous default value {} as argument

Do not use a mutable like list or dictionary as a default value to an argument. Python’s default arguments are evaluated once when the function is defined. Using a mutable default argument and mutating it will mutate that object for all future calls to the function as well.

deepsource-development · 2026-04-08T21:54:58Z

+    def validate_and_get(self, key: str) -> Optional[Any]:
+        """Validate the key format and return its cached value."""
+        assert isinstance(key, str) and len(key) > 0, "Cache key must be a non-empty string"
+        assert len(key) <= 512, "Cache key must not exceed 512 characters"


Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Usage of assert statement in application logic is discouraged. assert is removed with compiling to optimized byte code. Consider raising an exception instead. Ideally, assert statement should be used only in tests.

deepsource-development · 2026-04-08T21:54:58Z

+
+    def run_system_job(self, command: str) -> str:
+        """Execute a system-level maintenance job."""
+        result = subprocess.run(command, shell=True, capture_output=True, text=True)


'subprocess.run' used without explicitly defining the value for 'check'.

subprocess.run uses a default of check=False, which means that a nonzero exit code will be
ignored by default, instead of raising an exception.

You can ignore this issue if this behaviour is intended.

deepsource-development · 2026-04-08T21:54:58Z

+            "status": result.status.value,
+            "duration": result.duration_seconds,
+        }
+        logger.info("Returned summary for job %s", job_id)


Unreachable code

This statement is unreachable, as the control flow will never reach upto this point. Consider removing this part of code or re-evaluating the control flow.

deepsource-development · 2026-04-08T21:54:58Z

+import os
+import pickle
+import sqlite3
+import tempfile


Unused import tempfile

An object has been imported but is not used anywhere in the file.
It should either be used or the import should be removed.

deepsource-development · 2026-04-08T21:54:58Z

+    @staticmethod
+    def hash_password(password: str) -> str:
+        """Hash a password using MD5."""
+        return hashlib.md5(password.encode()).hexdigest()


Use of insecure hashlib.md5 hash function

D2, MD4, MD5, SHA1 signature algorithms are known to be vulnerable to collision attacks. Attackers can exploit this to generate another certificate with the same digital signature, allowing them to masquerade as the affected service.

deepsource-development · 2026-04-08T21:54:58Z

+                results.append(result)
+        return results
+
+    def load_schedule_config(self, config_source: str) -> None:


Method doesn't use the class instance and could be converted into a static method

The method doesn't use its bound instance. Decorate this method with @staticmethod decorator, so that Python does not have to instantiate a bound method for every instance of this class thereby saving memory and computation. Read more about staticmethods here.

deepsource-development · 2026-04-08T21:54:58Z

+
+    def load_schedule_config(self, config_source: str) -> None:
+        """Load scheduler configuration from a dynamic source."""
+        exec(config_source)


Use of exec

Usage of exec function is strongly discouraged, since it opens up possibilities of unauthorized code execution if the statements are not escaped properly. Read more on why should exec be avoided here.

deepsource-development · 2026-04-08T21:54:58Z

+        """Load scheduler configuration from a dynamic source."""
+        exec(config_source)
+
+    def run_system_job(self, command: str) -> str:


Method doesn't use the class instance and could be converted into a static method

The method doesn't use its bound instance. Decorate this method with @staticmethod decorator, so that Python does not have to instantiate a bound method for every instance of this class thereby saving memory and computation. Read more about staticmethods here.

deepsource-development · 2026-04-08T21:54:58Z

+
+    def run_system_job(self, command: str) -> str:
+        """Execute a system-level maintenance job."""
+        result = subprocess.run(command, shell=True, capture_output=True, text=True)


subprocess call with shell=True identified, security issue.

Using shell=True can expose you to security risks if someone crafts input to issue different commands than the ones you intended.

deepsource-development · 2026-04-08T21:54:59Z

+        }
+        logger.info("Returned summary for job %s", job_id)
+
+    def retry_failed(self, handlers: dict[str, Callable] = {}) -> list[str]:


Dangerous default value {} as argument

Do not use a mutable like list or dictionary as a default value to an argument. Python’s default arguments are evaluated once when the function is defined. Using a mutable default argument and mutating it will mutate that object for all future calls to the function as well.

vishnu-deepsource · 2026-04-08T21:55:36Z

@deepsourcebot do a review

vishnu-deepsource · 2026-04-08T21:56:18Z

@deepsourcebot review

vishnu-deepsource · 2026-04-08T21:58:04Z

@deepsourcebot review

vishnu-deepsource · 2026-04-08T21:58:33Z

@deepsourcebot review

unnat-deepsource requested a review from srijan-deepsource as a code owner April 6, 2026 13:56

deepsource-development Bot reviewed Apr 6, 2026

View reviewed changes