diff --git a/.coverage b/.coverage index 89f54e53..f626ef2c 100644 Binary files a/.coverage and b/.coverage differ diff --git a/src/qwed_new/api/main.py b/src/qwed_new/api/main.py index 67b3162c..afd8fbfc 100644 --- a/src/qwed_new/api/main.py +++ b/src/qwed_new/api/main.py @@ -457,7 +457,7 @@ async def verify_math( try: import sympy - from sympy.parsing.sympy_parser import parse_expr + from qwed_new.core.safe_parser import safe_parse_expr from sympy import simplify, symbols, Eq, solve expression = request.get("expression") @@ -472,8 +472,8 @@ async def verify_math( left_str, right_str = expression.split("=", 1) # Parse both sides - left = parse_expr(left_str) - right = parse_expr(right_str) + left = safe_parse_expr(left_str) + right = safe_parse_expr(right_str) # Simplify and check equivalence difference = simplify(left - right) @@ -501,7 +501,7 @@ async def verify_math( if re.search(r'/\d+\(', expression.replace(" ", "")): is_ambiguous = True - parsed = parse_expr(expression_normalized) + parsed = safe_parse_expr(expression_normalized) # Check for division by zero before simplifying if "/0" in expression.replace(" ", "") or "/ 0" in expression: diff --git a/src/qwed_new/core/batch.py b/src/qwed_new/core/batch.py index 666926d7..715cbe9a 100644 --- a/src/qwed_new/core/batch.py +++ b/src/qwed_new/core/batch.py @@ -220,14 +220,14 @@ async def _verify_item( ) elif item.verification_type == VerificationType.MATH: - from sympy.parsing.sympy_parser import parse_expr + from qwed_new.core.safe_parser import safe_parse_expr from sympy import simplify expression = item.query if "=" in expression: left, right = expression.split("=", 1) - left_expr = parse_expr(left) - right_expr = parse_expr(right) + left_expr = safe_parse_expr(left) + right_expr = safe_parse_expr(right) diff = simplify(left_expr - right_expr) is_valid = diff == 0 return { @@ -236,7 +236,7 @@ async def _verify_item( "message": "Identity verified" if is_valid else "Not equal" } else: - parsed = parse_expr(expression) + parsed = safe_parse_expr(expression) simplified = simplify(parsed) return { "is_valid": False, diff --git a/src/qwed_new/core/safe_parser.py b/src/qwed_new/core/safe_parser.py new file mode 100644 index 00000000..2bd1e7fe --- /dev/null +++ b/src/qwed_new/core/safe_parser.py @@ -0,0 +1,252 @@ +""" +Safe SymPy expression parser. + +Wraps sympy.parsing.sympy_parser.parse_expr with input validation, +a denylist for dangerous constructs, and a restricted evaluation +namespace. This module is the ONLY approved entry point for parsing +user-supplied math expressions in production code. + +Security boundary: + 1. Reject known-dangerous Python/OS constructs (denylist). + 2. Remove __builtins__ from the eval global dict. + 3. Allow-list only expected math symbols, constants, and functions. + 4. Enforce basic input validation (type, length, empty check). + +CWE-95 mitigation -- see PR #200 for full security analysis. +""" + +import ast +import re +from typing import Any, Dict, Optional, Tuple + +import sympy +from sympy import ( + E, I, Integer, Float, Rational, Symbol, oo, pi, +) +from sympy.parsing.sympy_parser import ( + parse_expr, + standard_transformations, + implicit_multiplication_application, +) + +__all__ = ["safe_parse_expr", "validate_variable_name", "get_safe_symbol", "SafeParserError"] + +MAX_EXPRESSION_LENGTH = 5_000 +_AST_MAX_DEPTH = 30 + +_DENYLIST_PATTERN = re.compile( + r"(?:" + r"__import__|__builtins__|__subclasses__|__globals__|__locals__" + r"|__getattr__|__setattr__|__delattr__|__class__|__bases__|__mro__" + r"|\beval\b|\bexec\b|\bcompile\b|\bgetattr\b|\bsetattr\b|\bdelattr\b" + r"|\bimport\b|\bimportlib\b" + r"|\bos\b|\bsys\b|\bsubprocess\b|\bshutil\b|\bsocket\b" + r"|\bpopen\b|\bsystem\b|\bspawn\b" + r"|\bopen\b|\bfile\b|\bpath\b|\bglob\b" + r"|\bchr\b|\bord\b|\bhex\b|\btype\b|\bvars\b|\bdir\b|\brepr\b" + r"|\binput\b|\bprint\b|\bbreakpoint\b|\bexit\b|\bquit\b" + r"|\bcodecs\b|\bcode\b|\bctypes\b" + r")", + re.IGNORECASE, +) + +_SAFE_GLOBAL_DICT_TEMPLATE: Dict[str, Any] = {"__builtins__": {}} + + +def _check_ast_depth(expression: str) -> None: + """Reject Python-parseable expressions exceeding max AST depth (DoS defence). + + Expressions using implicit multiplication (e.g. 2x, sin x) fail ast.parse + and skip this check — they are caught by the post-parse sympy depth check. + """ + try: + tree = ast.parse(expression, mode="eval") + except SyntaxError: + return + depth = _ast_node_depth(tree) + if depth > _AST_MAX_DEPTH: + raise SafeParserError( + f"Expression AST depth {depth} exceeds limit of {_AST_MAX_DEPTH}" + ) + + +def _ast_node_depth(node: ast.AST, current: int = 0) -> int: + max_depth = current + for child in ast.iter_child_nodes(node): + child_depth = _ast_node_depth(child, current + 1) + if child_depth > max_depth: + max_depth = child_depth + return max_depth + + +_SYMPY_MAX_DEPTH = 40 + + +def _sympy_tree_depth(expr: Any, current: int = 0) -> int: + """Compute nesting depth of a SymPy expression tree.""" + max_depth = current + for arg in getattr(expr, "args", ()): + child_depth = _sympy_tree_depth(arg, current + 1) + if child_depth > max_depth: + max_depth = child_depth + return max_depth + + +def _validate_sympy_result(result: Any) -> None: + """Ensure parse_expr returned a valid SymPy expression within depth limits.""" + import sympy + if not isinstance(result, sympy.Expr): + raise SafeParserError( + f"Parsed result is not a supported arithmetic expression, got {type(result).__name__}" + ) + depth = _sympy_tree_depth(result) + if depth > _SYMPY_MAX_DEPTH: + raise SafeParserError( + f"Expression tree depth {depth} exceeds limit of {_SYMPY_MAX_DEPTH}" + ) + + +def _build_safe_local_dict( + extra_symbols: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + safe: Dict[str, Any] = { + "x": Symbol("x"), "y": Symbol("y"), "z": Symbol("z"), + "a": Symbol("a"), "b": Symbol("b"), "c": Symbol("c"), + "d": Symbol("d"), "f": Symbol("f"), "g": Symbol("g"), + "h": Symbol("h"), "k": Symbol("k"), "m": Symbol("m"), + "n": Symbol("n", integer=True, positive=True), + "p": Symbol("p"), "q": Symbol("q"), "r": Symbol("r"), + "s": Symbol("s"), "t": Symbol("t"), "u": Symbol("u"), + "v": Symbol("v"), "w": Symbol("w"), + "alpha": Symbol("alpha"), "beta": Symbol("beta"), + "gamma": Symbol("gamma"), "delta": Symbol("delta"), + "epsilon": Symbol("epsilon"), "zeta": Symbol("zeta"), + "eta": Symbol("eta"), "theta": Symbol("theta"), + "iota": Symbol("iota"), "kappa": Symbol("kappa"), + "mu": Symbol("mu"), "nu": Symbol("nu"), + "xi": Symbol("xi"), "omicron": Symbol("omicron"), + "rho": Symbol("rho"), "sigma": Symbol("sigma"), + "tau": Symbol("tau"), "phi": Symbol("phi"), + "chi": Symbol("chi"), "psi": Symbol("psi"), + "omega": Symbol("omega"), + "pi": pi, "E": E, "I": I, "oo": oo, + "sin": sympy.sin, "cos": sympy.cos, "tan": sympy.tan, + "cot": sympy.cot, "sec": sympy.sec, "csc": sympy.csc, + "asin": sympy.asin, "acos": sympy.acos, "atan": sympy.atan, + "atan2": sympy.atan2, + "sinh": sympy.sinh, "cosh": sympy.cosh, "tanh": sympy.tanh, + "log": sympy.log, "ln": sympy.log, "exp": sympy.exp, + "sqrt": sympy.sqrt, "cbrt": sympy.cbrt, + "abs": sympy.Abs, "Abs": sympy.Abs, + "factorial": sympy.factorial, "binomial": sympy.binomial, + "Integer": Integer, "Float": Float, "Rational": Rational, + # Symbol is required because SymPy standard_transformations may emit + # Symbol('name') during evaluation. This allows users to create symbols + # with arbitrary names — the denylist and stripped builtins mitigate + # downstream attribute-access risks on resulting objects. + "Symbol": Symbol, + } + if extra_symbols: + for key, value in extra_symbols.items(): + if not isinstance(key, str): + raise SafeParserError( + f"extra_symbols keys must be strings, got {type(key).__name__}" + ) + if _DENYLIST_PATTERN.search(key): + raise SafeParserError( + f"extra_symbols key {key!r} contains disallowed construct" + ) + if not isinstance(value, (Symbol, sympy.Basic)): + raise SafeParserError( + f"extra_symbols[{key!r}] must be a SymPy Symbol or Basic, " + f"got {type(value).__name__}" + ) + safe[key] = value + return safe + + +class SafeParserError(ValueError): + pass + + +def safe_parse_expr( + expression: str, + *, + extra_symbols: Optional[Dict[str, Any]] = None, + transformations: Optional[Tuple] = None, +) -> Any: + if not isinstance(expression, str): + raise SafeParserError( + f"Expression must be a string, got {type(expression).__name__}" + ) + stripped = expression.strip() + if not stripped: + raise SafeParserError("Expression is empty") + if len(stripped) > MAX_EXPRESSION_LENGTH: + raise SafeParserError( + f"Expression exceeds maximum length of {MAX_EXPRESSION_LENGTH} characters" + ) + match = _DENYLIST_PATTERN.search(stripped) + if match: + raise SafeParserError( + f"Expression contains disallowed construct: {match.group()!r}" + ) + _check_ast_depth(stripped) + local_dict = _build_safe_local_dict(extra_symbols) + if transformations is None: + transformations = standard_transformations + ( + implicit_multiplication_application, + ) + global_dict = dict(_SAFE_GLOBAL_DICT_TEMPLATE) + try: + result = parse_expr( + stripped, + local_dict=local_dict, + global_dict=global_dict, + transformations=transformations, + ) + _validate_sympy_result(result) + return result + except SafeParserError: + raise + except Exception as exc: + raise SafeParserError(f"Failed to parse expression: {exc}") from exc + + +def validate_variable_name(variable: str) -> str: + if not isinstance(variable, str): + raise SafeParserError( + f"Variable name must be a string, got {type(variable).__name__}" + ) + stripped = variable.strip() + if not stripped: + raise SafeParserError("Variable name is empty") + if len(stripped) > 50: + raise SafeParserError("Variable name is too long") + if not re.match(r"^[A-Za-z][A-Za-z0-9_]*$", stripped): + raise SafeParserError( + f"Invalid variable name: {stripped!r}. " + "Must start with a letter and contain only alphanumeric characters." + ) + match = _DENYLIST_PATTERN.search(stripped) + if match: + raise SafeParserError( + f"Variable name contains disallowed construct: {match.group()!r}" + ) + return stripped + + +def get_safe_symbol(name: str) -> Symbol: + """Return a Symbol consistent with safe_parse_expr's namespace. + + Ensures calculus operation variables match any special assumptions + (e.g. Symbol(\"n\", integer=True, positive=True)) applied during parsing, + preventing symbol mismatch in diff/integrate/limit. + """ + name = validate_variable_name(name) + safe = _build_safe_local_dict() + if name in safe: + sym = safe[name] + if isinstance(sym, Symbol): + return sym + return Symbol(name) diff --git a/src/qwed_new/core/validator.py b/src/qwed_new/core/validator.py index 9fe93035..ed68460e 100644 --- a/src/qwed_new/core/validator.py +++ b/src/qwed_new/core/validator.py @@ -10,7 +10,7 @@ 3. Evaluable: Can we calculate a numerical result? """ -from sympy.parsing.sympy_parser import parse_expr +from qwed_new.core.safe_parser import safe_parse_expr from typing import Dict @@ -61,7 +61,7 @@ def validate(self, expression: str) -> Dict[str, any]: # Check 1: Syntax validation try: - expr = parse_expr(expression) + expr = safe_parse_expr(expression) checks_passed.append("syntax") except Exception as e: checks_failed.append("syntax") diff --git a/src/qwed_new/core/verifier.py b/src/qwed_new/core/verifier.py index 6f887b87..8d5a0bf0 100644 --- a/src/qwed_new/core/verifier.py +++ b/src/qwed_new/core/verifier.py @@ -18,7 +18,8 @@ diff, integrate, limit, oo, simplify, expand ) -from sympy.parsing.sympy_parser import parse_expr, standard_transformations, implicit_multiplication_application +from qwed_new.core.safe_parser import safe_parse_expr, get_safe_symbol +from sympy.parsing.sympy_parser import standard_transformations, implicit_multiplication_application from typing import Any, Dict, List, Optional from decimal import Decimal, ROUND_HALF_UP from dataclasses import dataclass @@ -120,7 +121,7 @@ def verify_math( try: # 1. Parse the expression safely - expr = parse_expr(expression, transformations=self.TRANSFORMATIONS) + expr = safe_parse_expr(expression) # 2. Evaluate deterministically if use_decimal: @@ -197,8 +198,8 @@ def verify_identity(self, lhs: str, rhs: str) -> Dict[str, Any]: verify_identity("sin(x)**2 + cos(x)**2", "1") # True """ try: - left = parse_expr(lhs, transformations=self.TRANSFORMATIONS) - right = parse_expr(rhs, transformations=self.TRANSFORMATIONS) + left = safe_parse_expr(lhs) + right = safe_parse_expr(rhs) # Method 1: Simplify difference diff = simplify(left - right) @@ -281,9 +282,9 @@ def verify_derivative( order: Order of derivative (1 for first, 2 for second, etc.) """ try: - expr = parse_expr(expression, transformations=self.TRANSFORMATIONS) - var = Symbol(variable) - expected_expr = parse_expr(expected, transformations=self.TRANSFORMATIONS) + expr = safe_parse_expr(expression) + var = get_safe_symbol(variable) + expected_expr = safe_parse_expr(expected) # Calculate derivative actual_derivative = diff(expr, var, order) @@ -328,14 +329,14 @@ def verify_integral( upper_bound: Upper bound for definite integral """ try: - expr = parse_expr(expression, transformations=self.TRANSFORMATIONS) - var = Symbol(variable) - expected_expr = parse_expr(expected, transformations=self.TRANSFORMATIONS) + expr = safe_parse_expr(expression) + var = get_safe_symbol(variable) + expected_expr = safe_parse_expr(expected) if lower_bound is not None and upper_bound is not None: # Definite integral - lower = parse_expr(lower_bound, transformations=self.TRANSFORMATIONS) - upper = parse_expr(upper_bound, transformations=self.TRANSFORMATIONS) + lower = safe_parse_expr(lower_bound) + upper = safe_parse_expr(upper_bound) actual_integral = integrate(expr, (var, lower, upper)) # For definite integrals, compare values @@ -385,9 +386,9 @@ def verify_limit( direction: "+" for right, "-" for left, "+-" for both """ try: - expr = parse_expr(expression, transformations=self.TRANSFORMATIONS) - var = Symbol(variable) - expected_expr = parse_expr(expected, transformations=self.TRANSFORMATIONS) + expr = safe_parse_expr(expression) + var = get_safe_symbol(variable) + expected_expr = safe_parse_expr(expected) # Parse the point (handle infinity) if point.lower() in ['oo', 'inf', 'infinity']: @@ -395,7 +396,7 @@ def verify_limit( elif point.lower() in ['-oo', '-inf', '-infinity']: pt = -oo else: - pt = parse_expr(point, transformations=self.TRANSFORMATIONS) + pt = safe_parse_expr(point) # Calculate limit actual_limit = limit(expr, var, pt, dir=direction) diff --git a/tests/security/test_safe_parser.py b/tests/security/test_safe_parser.py new file mode 100644 index 00000000..d6a2d344 --- /dev/null +++ b/tests/security/test_safe_parser.py @@ -0,0 +1,207 @@ +""" +Security tests for safe_parse_expr - CWE-95 mitigation. +""" + +import pytest +from qwed_new.core.safe_parser import ( + safe_parse_expr, + validate_variable_name, + SafeParserError, + MAX_EXPRESSION_LENGTH, +) + + +class TestDenylist: + + @pytest.mark.parametrize("payload", [ + '__import__("os").system("id")', + '__import__(chr(111)+chr(115)).system(chr(105)+chr(100))', + 'eval("1+1")', + 'exec("print(1)")', + 'os.system("id")', + 'os.popen("id").read()', + 'subprocess.call(["id"])', + '__builtins__["__import__"]("os")', + 'getattr(__import__("os"), "system")("id")', + 'open("/etc/passwd")', + '"".__class__.__bases__[0].__subclasses__()', + 'sys.modules["os"]', + 'compile("import os", "", "exec")', + 'chr(65)', + 'ord("A")', + 'print("hello")', + 'input("prompt")', + 'breakpoint()', + 'exit(0)', + 'type("X", (), {})', + 'dir()', + 'vars()', + ]) + def test_injection_blocked(self, payload): + with pytest.raises(SafeParserError, match="disallowed construct"): + safe_parse_expr(payload) + + +class TestLegitimateExpressions: + + @pytest.mark.parametrize("expr,expected_str", [ + ("2 + 2", "4"), + ("3 * 4 + 1", "13"), + ("10 / 2", "5"), + ("x**2 + 2*x + 1", "x**2 + 2*x + 1"), + ("(x + 1)**2", "(x + 1)**2"), + ("sin(pi/2)", "1"), + ("cos(0)", "1"), + ("log(1)", "0"), + ("exp(0)", "1"), + ("sqrt(16)", "4"), + ("factorial(5)", "120"), + ("pi", "pi"), + ("E", "E"), + ("Abs(-5)", "5"), + ]) + def test_valid_expression(self, expr, expected_str): + result = safe_parse_expr(expr) + assert str(result) == expected_str + + +class TestMultiLetterVariables: + + @pytest.mark.parametrize("var_name", [ + "alpha", "beta", "gamma", "delta", "epsilon", + "theta", "phi", "psi", "omega", "sigma", + "tau", "mu", "nu", "xi", "eta", "zeta", + "kappa", "rho", "chi", + ]) + def test_greek_variable(self, var_name): + result = safe_parse_expr(f"{var_name}**2 + 1") + assert var_name in str(result) + + def test_multi_variable_expression(self): + result = safe_parse_expr("alpha * beta + gamma") + result_str = str(result) + assert "alpha" in result_str + assert "beta" in result_str + assert "gamma" in result_str + + +class TestInputValidation: + + def test_non_string_input(self): + with pytest.raises(SafeParserError, match="must be a string"): + safe_parse_expr(42) + + def test_none_input(self): + with pytest.raises(SafeParserError, match="must be a string"): + safe_parse_expr(None) + + def test_empty_string(self): + with pytest.raises(SafeParserError, match="empty"): + safe_parse_expr("") + + def test_whitespace_only(self): + with pytest.raises(SafeParserError, match="empty"): + safe_parse_expr(" ") + + def test_too_long(self): + expr = "x + " * (MAX_EXPRESSION_LENGTH // 4 + 1) + with pytest.raises(SafeParserError, match="maximum length"): + safe_parse_expr(expr) + + +class TestVariableNameValidation: + + @pytest.mark.parametrize("name", ["x", "y", "theta", "alpha", "x1"]) + def test_valid_variable(self, name): + assert validate_variable_name(name) == name + + def test_non_string(self): + with pytest.raises(SafeParserError, match="must be a string"): + validate_variable_name(42) + + def test_empty(self): + with pytest.raises(SafeParserError, match="empty"): + validate_variable_name("") + + def test_too_long(self): + with pytest.raises(SafeParserError, match="too long"): + validate_variable_name("a" * 51) + + def test_starts_with_number(self): + with pytest.raises(SafeParserError, match="Invalid variable name"): + validate_variable_name("1x") + + def test_special_characters(self): + with pytest.raises(SafeParserError, match="Invalid variable name"): + validate_variable_name("x;drop") + + def test_denylist_blocked(self): + with pytest.raises(SafeParserError): + validate_variable_name("__import__") + + def test_os_blocked(self): + with pytest.raises(SafeParserError, match="disallowed"): + validate_variable_name("os") + + +class TestNamespaceIsolation: + + def test_global_dict_not_polluted(self): + safe_parse_expr("x + 1") + safe_parse_expr("y + 2") + result = safe_parse_expr("x + y") + assert "x" in str(result) + assert "y" in str(result) + + def test_global_dict_not_shared_between_calls(self): + from qwed_new.core.safe_parser import _SAFE_GLOBAL_DICT_TEMPLATE + before = dict(_SAFE_GLOBAL_DICT_TEMPLATE) + safe_parse_expr("x + 1") + safe_parse_expr("y + 2") + assert _SAFE_GLOBAL_DICT_TEMPLATE == before + + +class TestGetSafeSymbol: + + def test_consistency_with_n(self): + from qwed_new.core.safe_parser import get_safe_symbol + n1 = get_safe_symbol("n") + n2 = get_safe_symbol("n") + assert n1 == n2 + + def test_plain_symbol(self): + from qwed_new.core.safe_parser import get_safe_symbol + s = get_safe_symbol("myvar") + assert str(s) == "myvar" + + def test_invalid_variable(self): + from qwed_new.core.safe_parser import get_safe_symbol + with pytest.raises(SafeParserError): + get_safe_symbol("os.system") + + +class TestASTDepthLimit: + + def test_deeply_nested_rejected(self): + deep_expr = "x" + for _ in range(40): + deep_expr = f"sin({deep_expr})" + with pytest.raises(SafeParserError): + safe_parse_expr(deep_expr) + + def test_normal_expression_accepted(self): + result = safe_parse_expr("sin(cos(x + 1) * 2)") + assert result is not None + + +class TestExtraSymbols: + + def test_extra_symbol(self): + from sympy import Symbol + custom = {"myvar": Symbol("myvar")} + result = safe_parse_expr("myvar + 1", extra_symbols=custom) + assert "myvar" in str(result) + + def test_extra_symbol_rejects_non_sympy(self): + with pytest.raises(SafeParserError, match="must be a SymPy"): + safe_parse_expr("x + 1", extra_symbols={"bad": lambda: None}) diff --git a/tests/test_api_exceptions.py b/tests/test_api_exceptions.py index e02f7067..006ee364 100644 --- a/tests/test_api_exceptions.py +++ b/tests/test_api_exceptions.py @@ -31,7 +31,7 @@ def test_verify_math_exception_handling(): Test that verify_math catches internal errors and returns a sanitized message. """ # Force an exception during parsing/processing - with patch('sympy.parsing.sympy_parser.parse_expr', side_effect=ValueError("CRITICAL SENSITIVE STACK TRACE")): + with patch('qwed_new.core.safe_parser.parse_expr', side_effect=ValueError("CRITICAL SENSITIVE STACK TRACE")): # Providing valid minimal input to pass Pydantic validation response = client.post( "/verify/math",