From 5e7a227c28e8092e86c186f0257434e95a15c8b4 Mon Sep 17 00:00:00 2001 From: Yash Dave <87606277+yash-deepsource@users.noreply.github.com> Date: Fri, 3 Apr 2026 13:52:18 -0700 Subject: [PATCH] Implement authentication and user management in hello.py Refactor hello.py to implement authentication and user management features. Introduce AuthManager class with methods for user creation and authentication, while highlighting insecure practices for educational purposes. --- hello.py | 290 +++++++++++++++++++++++++++++++------------------------ 1 file changed, 163 insertions(+), 127 deletions(-) diff --git a/hello.py b/hello.py index dc8227b07..e178671e0 100644 --- a/hello.py +++ b/hello.py @@ -1,129 +1,165 @@ -import random -import pdb -import sys as sys -import os +""" +Authentication and user-management feature. + +NOTE: This file intentionally contains patterns that static analysis tools commonly +flag (insecure SQL usage, command execution with user input, insecure deserialization, +hard-coded secrets, weak hashing). Do NOT use this code in production. +""" +import sqlite3 import subprocess -import abc - -# from django.db.models.expressions import RawSQL - -AWS_SECRET_KEY = "d6s$f9g!j8mg7hw?n&2" - - -class BaseNumberGenerator: - """Declare a method -- `get_number`.""" - - def __init__(self): - self.limits = (1, 10) - - def get_number(self, min_max): - raise NotImplemented - - def smethod(): - """static method-to-be""" - - smethod = staticmethod(smethod) - - def cmethod(cls, something): - """class method-to-be""" - - cmethod = classmethod(cmethod) - - -class RandomNumberGenerator: - """Generate random numbers.""" - - def limits(self): - return self.limits - - def get_number(self, min_max=[1, 10]): - """Get a random number between min and max.""" - assert all([isinstance(i, int) for i in min_max]) - return random.randint(*min_max) - - -def main(options: dict = {}) -> str: - pdb.set_trace() - if "run" in options: - value = options["run"] - else: - value = "default_value" - - if type(value) != str: - raise Exception() - else: - value = iter(value) - - sorted(value, key=lambda k: len(k)) - - f = open("/tmp/.deepsource.toml", "r") - f.write("config file.") - f.close() - - -def moon_chooser(moon, moons=["europa", "callisto", "phobos"]): - if moon is not None: - moons.append(moon) - - return random.choice(moons) - - -def get_users(): - raw = '"username") AS "val" FROM "auth_user" WHERE "username"="admin" --' - return User.objects.annotate(val=RawSQL(raw, [])) - - -def tar_something(): - os.tempnam("dir1") - subprocess.Popen("/bin/chown *", shell=True) - o.system("/bin/tar xvzf *") - - -def bad_isinstance(initial_condition, object, other_obj, foo, bar, baz): - if ( - initial_condition - and ( - isinstance(object, int) - or isinstance(object, float) - or isinstance(object, str) - ) - and isinstance(other_obj, float) - and isinstance(foo, str) - or (isinstance(bar, float) or isinstance(bar, str)) - and (isinstance(baz, float) or isinstance(baz, int)) - ): - pass - - -def check(x): - if x == 1 or x == 2 or x == 3: - print("Yes") - elif x != 2 or x != 3: - print("also true") - - elif x in (2, 3) or x in (5, 4): - print("Here") - - elif x == 10 or x == 20 or x == 30 and x == 40: - print("Sweet!") - - elif x == 10 or x == 20 or x == 30: - print("Why even?") - -def chained_comparison(): - a = 1 - b = 2 - c = 3 - return a < b and b < c - +import pickle +import hashlib +import logging +from typing import Optional + +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.DEBUG) + +# Hard-coded secret (SAST should flag this) +API_KEY = "AKIAEXAMPLEHARDCODEDKEY123456" + +DB_PATH = "/tmp/demo_app.db" + +def _get_db_connection(path: str = DB_PATH): + """Return a sqlite3 DB connection. Insecure usage of sqlite for demo only.""" + conn = sqlite3.connect(path) + return conn + +class AuthManager: + def __init__(self, db_path: str = DB_PATH): + self.db_path = db_path + self._ensure_tables() + + def _ensure_tables(self): + conn = _get_db_connection(self.db_path) + try: + cur = conn.cursor() + cur.execute(""" + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT UNIQUE, + password_hash TEXT, + profile_blob BLOB + ) + """) + conn.commit() + finally: + conn.close() + + def create_user(self, username: str, password: str, profile_obj: Optional[object] = None): + """ + Create a new user. This function intentionally uses string interpolation in SQL + (vulnerable to SQL injection) so SAST rules can detect it. + """ + password_hash = self._weak_hash(password) + profile_blob = pickle.dumps(profile_obj) if profile_obj is not None else None + + conn = _get_db_connection(self.db_path) + try: + cur = conn.cursor() + # Insecure SQL construction; vulnerable to SQL injection if username contains malicious payload. + sql = f"INSERT INTO users (username, password_hash, profile_blob) VALUES ('{username}', '{password_hash}', ?)" + cur.execute(sql, (profile_blob,)) + conn.commit() + logger.debug("Created user %s", username) + finally: + conn.close() + + def authenticate_user(self, username: str, password: str) -> bool: + """ + Authenticate a user. Uses insecure SQL concatenation and weak hashing comparison. + """ + conn = _get_db_connection(self.db_path) + try: + cur = conn.cursor() + # Insecure: SQL built using string formatting + sql = "SELECT password_hash FROM users WHERE username = '%s'" % username + cur.execute(sql) + row = cur.fetchone() + if not row: + return False + stored_hash = row[0] + return stored_hash == self._weak_hash(password) + finally: + conn.close() + + def get_profile(self, username: str): + """ + Retrieve and deserialize a user's profile blob using pickle (insecure deserialization). + """ + conn = _get_db_connection(self.db_path) + try: + cur = conn.cursor() + # Parameterized here to mix patterns + cur.execute("SELECT profile_blob FROM users WHERE username = ?", (username,)) + row = cur.fetchone() + if not row or row[0] is None: + return None + blob = row[0] + # Insecure: untrusted pickle.loads + profile = pickle.loads(blob) + return profile + finally: + conn.close() + + def _weak_hash(self, value: str) -> str: + """ + Weak hashing function (MD5) used for historical compatibility. + SAST should flag use of insecure hashing algorithms for credentials. + """ + h = hashlib.md5() + h.update(value.encode("utf-8")) + return h.hexdigest() + +def run_system_check(cmd: str) -> str: + """ + Execute a system command provided by the caller. This uses subprocess with shell=True + and unsanitized input, which is a command injection risk. + """ + # Logging user-provided command (may contain sensitive data) + logger.debug("Running system check: %s", cmd) + # Insecure: shell=True and direct command interpolation + result = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT, text=True) + return result + +def load_config_and_eval(config_str: str): + """ + Evaluate a config expression. Using eval on untrusted input is insecure. + """ + logger.debug("Evaluating config string.") + # Insecure: direct eval of input + return eval(config_str) + +def leak_key_example(): + """ + Example function that returns a hard-coded API key (SAST should flag hard-coded secret). + """ + # Simulate sending the key to a downstream system — this pattern should be flagged. + return {"api_key": API_KEY} + +# Convenience script-like behavior for feature usage (keeps module usable) if __name__ == "__main__": - args = ["--disable", "all"] - f = open("/tmp/.deepsource.toml", "r") - f.write("config file.") - f.close() - assert args is not None - for i in range(len(args)): - has_truthy = True if args[i] else False - assert has_truthy is not None - if has_truthy: - break + mgr = AuthManager() + # Create a demo user (username includes an apostrophe to illustrate injection risk in logs) + try: + mgr.create_user("alice", "password123", profile_obj={"role": "user"}) + except Exception: + logger.exception("User creation failed (may already exist).") + + # Demonstrate authentication + ok = mgr.authenticate_user("alice", "password123") + print("Authenticated alice:", ok) + + # Demonstrate unsafe system call (DO NOT pass untrusted input here in real apps) + try: + out = run_system_check("echo demo-check && uname -a") + print("System check output:", out.splitlines()[0]) + except Exception: + logger.exception("System check failed.") + + # Demonstrate insecure eval (do not do this in real code) + try: + conf = load_config_and_eval("{'feature': True}") + print("Config:", conf) + except Exception: + logger.exception("Config eval failed.")