Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 20 additions & 10 deletions calculator.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Calculator module for demo."""
import sqlite3
import subprocess


def add(a: int, b: int) -> int:
Expand All @@ -12,16 +13,6 @@ def divide(a: float, b: float) -> float:
return a / b


def get_user_score(db_path: str, username: str) -> int | None:
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
result = cursor.execute(
"SELECT score FROM users WHERE username = ?",
(username,),
).fetchone()
return result[0] if result else None


def subtract(a: int, b: int) -> int:
return a - b

Expand All @@ -30,6 +21,25 @@ def multiply(a: int, b: int) -> int:
return a * b


def get_user_score(db_path: str, username: str) -> int | None:
# resource leak — exception before close() leaves connection open
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# SQL injection — f-string lets caller inject arbitrary SQL
result = cursor.execute(
f"SELECT score FROM users WHERE username = '{username}'"
).fetchone()
Comment on lines +29 to +31

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SQL injection regression. The original function used a parameterized query with ? — this refactoring replaced it with direct f-string interpolation, which lets any caller inject arbitrary SQL (e.g. username = "' OR '1'='1"). Revert to the parameterized form.

Suggested change
result = cursor.execute(
f"SELECT score FROM users WHERE username = '{username}'"
).fetchone()
result = cursor.execute(
"SELECT score FROM users WHERE username = ?",
(username,),
).fetchone()
  • Mark as noise

conn.close()
return result[0] if result else None
Comment on lines +24 to +33

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Resource leak. The original code used with sqlite3.connect(...) as conn: which guarantees the connection is closed even if execute() raises. The new code calls conn.close() unconditionally at the end — any exception (network error, malformed query, etc.) skips that line and leaks the connection. Use a context manager.

Suggested change
def get_user_score(db_path: str, username: str) -> int | None:
# resource leak — exception before close() leaves connection open
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# SQL injection — f-string lets caller inject arbitrary SQL
result = cursor.execute(
f"SELECT score FROM users WHERE username = '{username}'"
).fetchone()
conn.close()
return result[0] if result else None
def get_user_score(db_path: str, username: str) -> int | None:
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
result = cursor.execute(
"SELECT score FROM users WHERE username = ?",
(username,),
).fetchone()
return result[0] if result else None
  • Mark as noise



def run_report(report_name: str) -> str:
"""Generate a report by running an external tool."""
# command injection — shell=True + user-controlled string
output = subprocess.check_output(f"generate-report {report_name}", shell=True)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Command injection. shell=True with an unsanitized report_name lets callers execute arbitrary shell commands — e.g. report_name = "foo; rm -rf /". Pass the command as a list and drop shell=True.

Suggested change
output = subprocess.check_output(f"generate-report {report_name}", shell=True)
output = subprocess.check_output(["generate-report", report_name])
  • Mark as noise

return output.decode()


def process_scores(scores: list[int]) -> dict:
if not scores:
return {}
Expand Down
Loading