diff --git a/README.md b/README.md index 3a2a598..c820f8c 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,14 @@ at the call site, and a helper returning attacker-controlled data taints its res Each flow is then judged: a dominating guard that proves the value safe (`isdigit()`, membership in a constant allowlist, equality with a constant) refutes it and it is not reported; a guard that only mentions the value makes it a hotspot with medium confidence; -the verdict and its evidence are in the finding metadata. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing +the verdict and its evidence are in the finding metadata. Three more kinds of evidence +apply: a value proven numeric by the range analysis (`int()`, `len()`, arithmetic, +bounded by the comparisons on the path) cannot inject; a `Validator` model names a +callable whose truth proves one of its arguments, such as `re.fullmatch`; and an +`AuthorizationGuard` model names a decorator or a condition that restricts who reaches +the code, such as `login_required`, behind which a flow is a hotspot rather than a +vulnerability. Plugins contribute validators and authorization guards through their +models, like sources and sinks. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing was found, 1 when findings were reported, and 2 on a usage or analysis error. ## Emit PyIR diff --git a/plugins/models/django/django_models.py b/plugins/models/django/django_models.py index 5284883..565049e 100644 --- a/plugins/models/django/django_models.py +++ b/plugins/models/django/django_models.py @@ -13,7 +13,15 @@ from coretrace_python.plugins import ModelPlugin from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.taint import EntryPoint, Model, Sanitizer, Sink, TaintKind, TypedParameter +from coretrace_python.taint import ( + AuthorizationGuard, + EntryPoint, + Model, + Sanitizer, + Sink, + TaintKind, + TypedParameter, +) _REQUEST_CLASSES = ( "django.http.HttpRequest", @@ -59,6 +67,14 @@ "rest_framework.decorators.action", ) +_AUTHORIZATION_DECORATORS = ( + ("django.contrib.auth.decorators.login_required", "login"), + ("django.contrib.auth.decorators.permission_required", "permission"), + ("django.contrib.auth.decorators.user_passes_test", "test"), + ("django.contrib.admin.views.decorators.staff_member_required", "staff"), + ("rest_framework.decorators.permission_classes", "permission"), +) + def _sym(path: str) -> SymbolId: return SymbolId(f"python.{path}") @@ -83,4 +99,5 @@ class DjangoModels(ModelPlugin): Sink(_sym("django.http.FileResponse"), TaintKind.PATH), Sanitizer(_sym("django.utils.html.escape"), TaintKind.HTML), Sanitizer(_sym("django.utils.html.conditional_escape"), TaintKind.HTML), + *(AuthorizationGuard(_sym(decorator), label) for decorator, label in _AUTHORIZATION_DECORATORS), ) diff --git a/plugins/models/flask/flask_models.py b/plugins/models/flask/flask_models.py index d2faba3..bd7dd43 100644 --- a/plugins/models/flask/flask_models.py +++ b/plugins/models/flask/flask_models.py @@ -6,7 +6,15 @@ from coretrace_python.plugins import ModelPlugin from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.taint import EntryPoint, Model, Sanitizer, Sink, Source, TaintKind +from coretrace_python.taint import ( + AuthorizationGuard, + EntryPoint, + Model, + Sanitizer, + Sink, + Source, + TaintKind, +) _REQUEST_ATTRIBUTES = ( "args", "form", "values", "json", "data", "cookies", "headers", "files", @@ -31,4 +39,7 @@ class FlaskModels(ModelPlugin): Sink(_sym("flask.send_file"), TaintKind.PATH), Sanitizer(_sym("flask.escape"), TaintKind.HTML), Sanitizer(_sym("markupsafe.escape"), TaintKind.HTML), + AuthorizationGuard(_sym("flask_login.login_required"), "login"), + AuthorizationGuard(_sym("flask_login.fresh_login_required"), "login"), + AuthorizationGuard(_sym("flask_login.current_user.is_authenticated"), "login"), ) diff --git a/plugins/models/python_stdlib/python_stdlib.py b/plugins/models/python_stdlib/python_stdlib.py index 82a0447..2dd9018 100644 --- a/plugins/models/python_stdlib/python_stdlib.py +++ b/plugins/models/python_stdlib/python_stdlib.py @@ -6,7 +6,7 @@ from coretrace_python.plugins import ModelPlugin from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.taint import Model, Sanitizer, Sink, Source, TaintKind +from coretrace_python.taint import Model, Sanitizer, Sink, Source, TaintKind, Validator def _sym(path: str) -> SymbolId: @@ -43,4 +43,6 @@ class PythonStdlibModels(ModelPlugin): Sanitizer(_sym("shlex.quote"), TaintKind.COMMAND), Sanitizer(_sym("html.escape"), TaintKind.HTML), Sanitizer(_sym("os.path.basename"), TaintKind.PATH), + Validator(_sym("re.fullmatch"), argument=1), + Validator(_sym("re.compile.fullmatch")), ) diff --git a/src/coretrace_python/abstract/__init__.py b/src/coretrace_python/abstract/__init__.py index b0ee686..5e2233f 100644 --- a/src/coretrace_python/abstract/__init__.py +++ b/src/coretrace_python/abstract/__init__.py @@ -5,12 +5,17 @@ ConstantPropagation, propagate_constants, ) +from coretrace_python.abstract.ranges import Interval, RangeAnalysis, RangeFacts, analyze_ranges from coretrace_python.abstract.values import AbstractValue, Truth __all__ = [ "AbstractValue", "ConstantFacts", "ConstantPropagation", + "Interval", + "RangeAnalysis", + "RangeFacts", "Truth", + "analyze_ranges", "propagate_constants", ] diff --git a/src/coretrace_python/abstract/ranges.py b/src/coretrace_python/abstract/ranges.py new file mode 100644 index 0000000..7237756 --- /dev/null +++ b/src/coretrace_python/abstract/ranges.py @@ -0,0 +1,285 @@ +"""Numeric range analysis (architecture §18, §24). + +Every value proven numeric gets an ``Interval``: numeric constants, arithmetic on +numbers, the results of ``int()``, ``len()`` and friends. Comparisons refine the +intervals of both sides on the branch they take, chained comparisons and ``and`` / ``or`` +included, and loops converge by widening. Values not proven numeric have no interval: +the domain doubles as a proof of numeric type, which is what the refutation engine +needs, since a number cannot carry an injection. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType +from typing import ClassVar + +from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis +from coretrace_python.cfg import CFG, BlockId, CFGAnalysis +from coretrace_python.dataflow import DataflowProblem, Direction, solve +from coretrace_python.hir import nodes +from coretrace_python.ir.model import ( + BasicBlock, + BinaryOp, + BoolOp, + Branch, + Call, + Compare, + Constant, + ForNext, + FunctionIR, + Instruction, + Jump, + Phi, + Symbol, + UnaryOp, + Value, +) +from coretrace_python.ir.ssa import SSAAnalysis + +INF = float("inf") + + +def _format(bound: float) -> str: + if bound in (INF, -INF): + return "inf" if bound > 0 else "-inf" + return str(int(bound)) if float(bound).is_integer() else str(bound) + + +@dataclass(frozen=True) +class Interval: + """A closed range of numbers; ``inf`` bounds mean unbounded.""" + + low: float + high: float + + def join(self, other: Interval) -> Interval: + return Interval(min(self.low, other.low), max(self.high, other.high)) + + def widen(self, other: Interval) -> Interval: + """``self`` extended to ``other``, jumping to infinity on any side that grew.""" + + return Interval( + self.low if other.low >= self.low else -INF, + self.high if other.high <= self.high else INF, + ) + + def __str__(self) -> str: + return f"[{_format(self.low)}, {_format(self.high)}]" + + +UNBOUNDED = Interval(-INF, INF) +NON_NEGATIVE = Interval(0, INF) + +_NUMERIC_CALLS: Mapping[str, Interval] = { + "python.builtins.int": UNBOUNDED, + "python.builtins.float": UNBOUNDED, + "python.builtins.round": UNBOUNDED, + "python.builtins.len": NON_NEGATIVE, + "python.builtins.abs": NON_NEGATIVE, + "python.builtins.ord": Interval(0, 0x10FFFF), +} +_HULL_CALLS = frozenset({"python.builtins.min", "python.builtins.max"}) + +State = Mapping[Value, Interval] + + +def _times(a: float, b: float) -> float: + return 0.0 if a == 0 or b == 0 else a * b + + +class RangeFacts: + def __init__(self, states: Mapping[BlockId, State]) -> None: + self._states: Mapping[BlockId, State] = MappingProxyType( + {block: MappingProxyType(dict(state)) for block, state in states.items()} + ) + + def at(self, block: BlockId) -> State: + """The intervals known at the end of ``block``; empty when it is unreachable.""" + + return self._states.get(block, MappingProxyType({})) + + +class _RangeProblem(DataflowProblem[State]): + direction: ClassVar[Direction] = Direction.FORWARD + + def __init__(self, function: FunctionIR) -> None: + self.function = function + self.blocks = {block.id: block for block in function.blocks} + self.defs: dict[Value, Instruction] = { + i.result: i for block in function.blocks for i in block.instructions if i.result + } + self._widened: dict[Value, Interval] = {} + + def initial(self) -> State: + return MappingProxyType({}) + + def join(self, a: State, b: State) -> State: + return MappingProxyType( + {value: a[value].join(b[value]) for value in a if value in b} + ) + + def evaluate(self, block: BasicBlock, incoming: Mapping[BlockId, State]) -> State: + states = list(incoming.values()) + state: dict[Value, Interval] = dict(states[0]) if states else {} + for other in states[1:]: + state = dict(self.join(state, other)) + for instruction in block.instructions: + if instruction.result is None: + continue + found = ( + self.phi(instruction, incoming) + if isinstance(instruction, Phi) + else self.instruction(instruction, state) + ) + if found is None: + state.pop(instruction.result, None) + else: + state[instruction.result] = found + return MappingProxyType(state) + + def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) -> Mapping[BlockId, State]: + block = self.blocks[block_id] + state = self.evaluate(block, incoming) + exits = {target: state for target in block.exception_targets} + terminator = block.terminator + if isinstance(terminator, Branch): + return { + **exits, + terminator.then_block: self.refined(state, terminator.condition, True), + terminator.else_block: self.refined(state, terminator.condition, False), + } + if isinstance(terminator, Jump): + return {**exits, terminator.target: state} + if isinstance(terminator, ForNext): + return {**exits, terminator.body: state, terminator.exit: state} + return exits + + # ------------------------------------------------------------------ transfer + + def phi(self, phi: Phi, incoming: Mapping[BlockId, State]) -> Interval | None: + result: Interval | None = None + for value, predecessor in phi.incoming: + if predecessor not in incoming: + continue + found = incoming[predecessor].get(value) + if found is None: + return None + result = found if result is None else result.join(found) + if result is None: + return None + previous = self._widened.get(phi.result) + widened = result if previous is None else previous.widen(previous.join(result)) + self._widened[phi.result] = widened + return widened + + def instruction(self, instruction: Instruction, state: Mapping[Value, Interval]) -> Interval | None: + if isinstance(instruction, Constant): + value = instruction.value + if isinstance(value, bool | int | float) and value == value: # noqa: PLR0124 - NaN + return Interval(float(value), float(value)) + return None + if isinstance(instruction, BinaryOp): + left, right = state.get(instruction.left), state.get(instruction.right) + if left is None or right is None: + return None + return self.binary(instruction.operator, left, right) + if isinstance(instruction, UnaryOp): + operand = state.get(instruction.operand) + if instruction.operator == "not": + return Interval(0, 1) + if operand is None: + return None + if instruction.operator == "neg": + return Interval(-operand.high, -operand.low) + return operand if instruction.operator == "pos" else UNBOUNDED + if isinstance(instruction, Compare): + return Interval(0, 1) + if isinstance(instruction, Call): + callee = self.defs.get(instruction.callee) + if not isinstance(callee, Symbol): + return None + name = callee.symbol_id.canonical_name + if name in _NUMERIC_CALLS: + return _NUMERIC_CALLS[name] + if name in _HULL_CALLS and instruction.arguments and not instruction.keywords: + hull: Interval | None = None + for argument in instruction.arguments: + found = state.get(argument) + if found is None: + return None + hull = found if hull is None else hull.join(found) + return hull + return None + + @staticmethod + def binary(operator: str, left: Interval, right: Interval) -> Interval | None: + if operator == "add": + return Interval(left.low + right.low, left.high + right.high) + if operator == "sub": + return Interval(left.low - right.high, left.high - right.low) + if operator == "mul": + products = [ + _times(a, b) for a in (left.low, left.high) for b in (right.low, right.high) + ] + return Interval(min(products), max(products)) + if operator in ("div", "floor_div", "mod", "pow", "bit_or", "bit_xor", "bit_and", "lshift", "rshift"): + return UNBOUNDED + return None + + # ------------------------------------------------------------------ refinement + + def refined(self, state: State, condition: Value, truth: bool) -> State: + overrides: dict[Value, Interval] = {} + self.refine(dict(state), condition, truth, overrides) + return MappingProxyType({**state, **overrides}) if overrides else state + + def refine( + self, state: dict[Value, Interval], condition: Value, truth: bool, overrides: dict[Value, Interval] + ) -> None: + definition = self.defs.get(condition) + if isinstance(definition, UnaryOp) and definition.operator == "not": + self.refine(state, definition.operand, not truth, overrides) + elif isinstance(definition, BoolOp) and (definition.operator == "and") == truth: + for value in definition.values: + self.refine(state, value, truth, overrides) + elif isinstance(definition, Compare): + left, right = definition.left, definition.right + if left not in state or right not in state: + return + operator = definition.operator + if operator == "eq" and truth or operator == "not_eq" and not truth: + low = max(state[left].low, state[right].low) + high = min(state[left].high, state[right].high) + if low <= high: + overrides[left] = overrides[right] = Interval(low, high) + state[left] = state[right] = Interval(low, high) + return + if operator not in ("lt", "lt_eq", "gt", "gt_eq"): + return + ascending = (operator in ("lt", "lt_eq")) == truth + lower, upper = (left, right) if ascending else (right, left) + bounded_lower = Interval(state[lower].low, min(state[lower].high, state[upper].high)) + bounded_upper = Interval(max(state[upper].low, state[lower].low), state[upper].high) + overrides[lower], overrides[upper] = bounded_lower, bounded_upper + state[lower], state[upper] = bounded_lower, bounded_upper + + +def analyze_ranges(function: FunctionIR, cfg: CFG) -> RangeFacts: + problem = _RangeProblem(function) + solution = solve(problem, cfg) + states: dict[BlockId, State] = {} + for block in function.blocks: + if solution.reached(block.id): + states[block.id] = problem.evaluate(block, solution.incoming(block.id)) + return RangeFacts(states) + + +class RangeAnalysis(FunctionAnalysis[RangeFacts]): + name: ClassVar[str] = "abstract.ranges" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, CFGAnalysis}) + + @classmethod + def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> RangeFacts: + return analyze_ranges(ctx.get(SSAAnalysis, function), ctx.get(CFGAnalysis, function)) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index ed5378c..415a7fa 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -15,7 +15,7 @@ from typing import Any, ClassVar from coretrace_python import __version__ -from coretrace_python.abstract import ConstantPropagation +from coretrace_python.abstract import ConstantPropagation, RangeAnalysis from coretrace_python.analysis import ( AnalysisContext, AnalysisManager, @@ -101,6 +101,7 @@ SSAAnalysis, DefUseAnalysis, ConstantPropagation, + RangeAnalysis, CallGraphAnalysis, SummaryAnalysis, ProjectSummaries, diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py index 5deae47..3efff39 100644 --- a/src/coretrace_python/findings/refutation.py +++ b/src/coretrace_python/findings/refutation.py @@ -4,10 +4,14 @@ whose one side alone reaches the sink fixes the truth of its condition there; the condition is then interpreted: string validators (``isdigit()`` and friends), membership in a constant allowlist and equality with a constant prove a value safe, ``and`` / ``or`` -and ``not`` combine as expected, and anything else that mentions the value is a guard -that does not prove it. A flow is refuted when every tainted origin of the argument is -proven safe or the sink is unreachable, a hotspot when an unproven guard mentions it, and -a vulnerability otherwise. +and ``not`` combine as expected, a ``Validator`` model names a callable whose truth proves +one of its arguments, a numeric value (``abstract.ranges``) cannot inject, and anything +else that mentions the value is a guard that does not prove it. A proof counts for an +origin when every dependence path from that origin to the sink argument goes through a +proven value. A flow is refuted when every tainted origin is proven safe or the sink is +unreachable, a hotspot when it sits behind an ``AuthorizationGuard`` (a decorator or a +dominating condition) or when an unproven guard mentions it, and a vulnerability +otherwise. """ from __future__ import annotations @@ -18,10 +22,11 @@ from types import MappingProxyType from typing import ClassVar -from coretrace_python.abstract import ConstantPropagation +from coretrace_python.abstract import ConstantPropagation, RangeAnalysis, RangeFacts from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis from coretrace_python.cfg import CFG, BlockId, CFGAnalysis, DominanceAnalysis, DominatorTree from coretrace_python.hir import nodes +from coretrace_python.interprocedural import CallGraphAnalysis from coretrace_python.ir.model import ( BoolOp, Branch, @@ -37,7 +42,16 @@ Value, ) from coretrace_python.ir.ssa import SSAAnalysis -from coretrace_python.taint import TaintAnalysis, TaintFacts, TaintFlow +from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeTable +from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, SymbolTable +from coretrace_python.taint import ( + AuthorizationGuard, + ModelTable, + SecurityModelAnalysis, + TaintAnalysis, + TaintFacts, + TaintFlow, +) VALIDATORS = frozenset( {"isdigit", "isnumeric", "isdecimal", "isalnum", "isalpha", "isidentifier", "isascii"} @@ -82,12 +96,24 @@ class _Guard: class _Judge: def __init__( - self, function: FunctionIR, cfg: CFG, tree: DominatorTree, taint: TaintFacts + self, + function: FunctionIR, + cfg: CFG, + tree: DominatorTree, + taint: TaintFacts, + ranges: RangeFacts | None = None, + models: ModelTable | None = None, + symbols: Mapping[Value, SymbolId] | None = None, + authorization: AuthorizationGuard | None = None, ) -> None: self.function = function self.cfg = cfg self.tree = tree self.taint = taint + self.ranges = ranges or RangeFacts({}) + self.models = models or ModelTable((), (), ()) + self.symbols = symbols or {} + self.authorization = authorization self.blocks = {block.id: block for block in function.blocks} self.defs: dict[Value, Instruction] = { i.result: i for block in function.blocks for i in block.instructions if i.result @@ -125,6 +151,26 @@ def origins(self, value: Value) -> frozenset[Value]: and not any(self.taint.taint(o) for o in self.closure(v)) ) + def covered(self, origin: Value, argument: Value, proven: Mapping[Value, str]) -> bool: + """Whether every dependence path from ``origin`` to ``argument`` goes through a + proven value, so nothing of the origin reaches the sink unproven.""" + + if origin in proven: + return True + seen: set[Value] = set() + pending = [argument] + while pending: + value = pending.pop() + if value in proven or value in seen: + continue + if value == origin: + return False + seen.add(value) + definition = self.defs.get(value) + if definition is not None: + pending.extend(definition.operands()) + return True + # ------------------------------------------------------------------ guards def sink_block(self, flow: TaintFlow) -> BlockId | None: @@ -190,7 +236,9 @@ def interpret( if truth is None: mentioned = set(self.closure(condition)) | {condition} elif truth == when_true: + # The check proves ``tested``; whatever else it reads is merely mentioned. proven[tested] = reason + mentioned = set(self.closure(condition)) | {condition} return proven, mentioned def recognise(self, definition: Instruction | None) -> tuple[Value, str, bool] | None: @@ -200,6 +248,11 @@ def recognise(self, definition: Instruction | None) -> tuple[Value, str, bool] | callee = self.defs.get(definition.callee) if isinstance(callee, GetAttr) and callee.attribute in VALIDATORS: return callee.object, f"guarded by {callee.attribute}()", True + if isinstance(definition, Call): + symbol = self.symbols.get(definition.callee) + validator = self.models.validator(symbol) if symbol is not None else None + if validator is not None and validator.argument < len(definition.arguments): + return definition.arguments[validator.argument], f"validated by {symbol}", True if isinstance(definition, Compare): left, right = definition.left, definition.right if definition.operator in ("in", "not_in") and self.is_constant_collection(right): @@ -222,6 +275,27 @@ def is_constant_collection(self, value: Value) -> bool: return all(isinstance(self.defs.get(e), Constant) for e in definition.elements) return False + # ------------------------------------------------------------------ authorization + + def authorized_by(self, condition: Value, truth: bool) -> AuthorizationGuard | None: + """The authorization guard this condition enforces when it is ``truth``.""" + + definition = self.defs.get(condition) + if isinstance(definition, UnaryOp) and definition.operator == "not": + return self.authorized_by(definition.operand, not truth) + if isinstance(definition, BoolOp) and (definition.operator == "and") == truth: + for value in definition.values: + found = self.authorized_by(value, truth) + if found is not None: + return found + return None + if not truth: + return None + symbol = self.symbols.get(condition) + if symbol is None and isinstance(definition, Call): + symbol = self.symbols.get(definition.callee) + return self.models.authorization(symbol) if symbol is not None else None + # ------------------------------------------------------------------ verdicts def judge(self, flow: TaintFlow, reachable: bool) -> Verdict: @@ -229,25 +303,44 @@ def judge(self, flow: TaintFlow, reachable: bool) -> Verdict: if sink is None or not reachable: return Verdict(flow, Status.REFUTED, "sink unreachable by constant propagation") origins = self.origins(flow.argument) - chains = { - origin: frozenset( - w for w in self.closure(flow.argument) | {flow.argument} - if w == origin or origin in self.closure(w) - ) - for origin in origins + chain = self.closure(flow.argument) | {flow.argument} + proven: dict[Value, str] = { + value: f"numeric value within {interval}" + for value, interval in self.ranges.at(sink).items() + if value in chain } - proofs: dict[Value, str] = {} mentions: dict[Value, int] = {} + authorization: str | None = ( + f"behind authorization ({self.authorization.label}) by decorator" + if self.authorization is not None + else None + ) for guard in self.guards(sink): - proven, mentioned = self.interpret(guard.condition, guard.truth) - for origin, chain in chains.items(): - for value, reason in proven.items(): - if value in chain: - proofs.setdefault(origin, reason) - if origin not in mentions and mentioned & chain: + found, mentioned = self.interpret(guard.condition, guard.truth) + for value, reason in found.items(): + if value in chain: + proven.setdefault(value, reason) + for origin in origins: + if origin not in mentions and mentioned & ( + {origin} | {w for w in chain if origin in self.closure(w)} + ): mentions[origin] = guard.line + if authorization is None: + guard_model = self.authorized_by(guard.condition, guard.truth) + if guard_model is not None: + authorization = f"behind authorization ({guard_model.label}) at line {guard.line}" + proofs = { + origin: sorted( + {reason for value, reason in proven.items() if value == origin or origin in self.closure(value)} + ) + for origin in origins + if self.covered(origin, flow.argument, proven) + } if origins and all(origin in proofs for origin in origins): - return Verdict(flow, Status.REFUTED, "; ".join(sorted(set(proofs.values())))) + reasons = sorted({reason for found in proofs.values() for reason in found}) + return Verdict(flow, Status.REFUTED, "; ".join(reasons)) + if authorization is not None: + return Verdict(flow, Status.HOTSPOT, authorization) unguarded = [origin for origin in origins if origin not in proofs and origin not in mentions] if not origins or unguarded: return Verdict(flow, Status.VULNERABILITY, "no guard on the path to the sink") @@ -263,8 +356,12 @@ def judge_flows( tree: DominatorTree, taint: TaintFacts, reachable: frozenset[BlockId], + ranges: RangeFacts | None = None, + models: ModelTable | None = None, + symbols: Mapping[Value, SymbolId] | None = None, + authorization: AuthorizationGuard | None = None, ) -> Verdicts: - judge = _Judge(function, cfg, tree, taint) + judge = _Judge(function, cfg, tree, taint, ranges, models, symbols, authorization) verdicts = [] for flow in taint.flows: sink = judge.sink_block(flow) @@ -272,21 +369,53 @@ def judge_flows( return Verdicts(tuple(verdicts)) +def authorization_of( + function: nodes.Function, models: ModelTable, scopes: ScopeTable, symbols: SymbolTable +) -> AuthorizationGuard | None: + """The authorization guard among the function's decorators, if any.""" + + scope = scopes.scope_for(function) + enclosing = scope.parent if scope.parent is not None else scope.id + for decorator in function.decorators: + symbol = symbols.resolve_expression(enclosing, decorator) + guard = models.authorization(symbol) if symbol is not None else None + if guard is not None: + return guard + return None + + class RefutationAnalysis(FunctionAnalysis[Verdicts]): name: ClassVar[str] = "findings.refutation" requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( - {TaintAnalysis, DominanceAnalysis, ConstantPropagation, SSAAnalysis, CFGAnalysis} + { + TaintAnalysis, + DominanceAnalysis, + ConstantPropagation, + RangeAnalysis, + SSAAnalysis, + CFGAnalysis, + SecurityModelAnalysis, + CallGraphAnalysis, + ScopeAnalysis, + SymbolAnalysis, + } ) @classmethod def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> Verdicts: ssa = ctx.get(SSAAnalysis, function) constants = ctx.get(ConstantPropagation, function) + graph = ctx.get(CallGraphAnalysis) + models = ctx.get(SecurityModelAnalysis) return judge_flows( ssa, ctx.get(CFGAnalysis, function), ctx.get(DominanceAnalysis, function), ctx.get(TaintAnalysis, function), frozenset(b.id for b in ssa.blocks if constants.reachable(b.id)), + ctx.get(RangeAnalysis, function), + models, + graph.symbols(graph.name_of(function)), + authorization_of(function, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis)), ) diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index 23242a9..f5d2505 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/src/coretrace_python/taint/__init__.py @@ -8,6 +8,7 @@ propagate_taint, ) from coretrace_python.taint.models import ( + AuthorizationGuard, EntryPoint, Model, ModelError, @@ -19,9 +20,11 @@ Source, TaintKind, TypedParameter, + Validator, ) __all__ = [ + "AuthorizationGuard", "EntryPoint", "Model", "ModelError", @@ -37,5 +40,6 @@ "TaintFlow", "TaintKind", "TypedParameter", + "Validator", "propagate_taint", ] diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 538c118..9ccbf31 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -77,7 +77,25 @@ class TypedParameter: kinds: TaintKind = TaintKind.ALL -Model = Source | Sink | Sanitizer | EntryPoint | TypedParameter +@dataclass(frozen=True) +class Validator: + """A callable whose truth proves its ``argument`` safe (refutation evidence, §24).""" + + symbol: SymbolId + kinds: TaintKind = TaintKind.ALL + argument: int = 0 + + +@dataclass(frozen=True) +class AuthorizationGuard: + """A decorator, or a condition, that restricts who reaches the code behind it; a + flow behind one is a hotspot rather than a vulnerability (§24).""" + + symbol: SymbolId + label: str + + +Model = Source | Sink | Sanitizer | EntryPoint | TypedParameter | Validator | AuthorizationGuard @dataclass(frozen=True) @@ -87,6 +105,8 @@ class ModelTable: sanitizers: tuple[Sanitizer, ...] entry_points: tuple[EntryPoint, ...] = () typed_parameters: tuple[TypedParameter, ...] = () + validators: tuple[Validator, ...] = () + authorizations: tuple[AuthorizationGuard, ...] = () _by_symbol: dict[type[Model], dict[SymbolId, Model]] = field( init=False, repr=False, compare=False ) @@ -98,6 +118,8 @@ def __post_init__(self) -> None: Sanitizer: {m.symbol: m for m in self.sanitizers}, EntryPoint: {m.symbol: m for m in self.entry_points}, TypedParameter: {m.symbol: m for m in self.typed_parameters}, + Validator: {m.symbol: m for m in self.validators}, + AuthorizationGuard: {m.symbol: m for m in self.authorizations}, } object.__setattr__(self, "_by_symbol", MappingProxyType(index)) @@ -109,6 +131,14 @@ def typed_parameter(self, symbol: SymbolId) -> TypedParameter | None: found = self._by_symbol[TypedParameter].get(symbol) return found if isinstance(found, TypedParameter) else None + def validator(self, symbol: SymbolId) -> Validator | None: + found = self._by_symbol[Validator].get(symbol) + return found if isinstance(found, Validator) else None + + def authorization(self, symbol: SymbolId) -> AuthorizationGuard | None: + found = self._by_symbol[AuthorizationGuard].get(symbol) + return found if isinstance(found, AuthorizationGuard) else None + def source(self, symbol: SymbolId) -> Source | None: found = self._by_symbol[Source].get(symbol) return found if isinstance(found, Source) else None @@ -143,6 +173,8 @@ def extended(self, *sinks: Sink) -> ModelTable: self.sanitizers, self.entry_points, self.typed_parameters, + self.validators, + self.authorizations, ) def sanitizer(self, symbol: SymbolId) -> Sanitizer | None: @@ -173,6 +205,8 @@ def freeze(self) -> ModelTable: sanitizers=tuple(m for m in models if isinstance(m, Sanitizer)), entry_points=tuple(m for m in models if isinstance(m, EntryPoint)), typed_parameters=tuple(m for m in models if isinstance(m, TypedParameter)), + validators=tuple(m for m in models if isinstance(m, Validator)), + authorizations=tuple(m for m in models if isinstance(m, AuthorizationGuard)), ) diff --git a/tests/test_refutation_evidence.py b/tests/test_refutation_evidence.py new file mode 100644 index 0000000..86a09f1 --- /dev/null +++ b/tests/test_refutation_evidence.py @@ -0,0 +1,337 @@ +"""Acceptance tests for the remaining evidence of the proof engine (``docs/architecture.md`` +§24, §25 plugins/proof; roadmap issue #32). + +- Numeric ranges: ``abstract.ranges`` keeps an interval for every value proven numeric + (numeric constants, arithmetic on numbers, ``int()``, ``len()`` and friends), refined + on the branches of comparisons and widened over loops. A tainted origin whose every + dependence path to the sink argument goes through a numeric value cannot inject. +- Validators: a ``Validator`` model names a callable whose truth proves one of its + arguments safe, so ``re.fullmatch`` and framework validators refute like ``isdigit()``. +- Authorization: an ``AuthorizationGuard`` model names a decorator or a condition that + restricts who reaches the sink; a flow behind one is a hotspot, not a vulnerability. + +Plugins contribute both models through ``Plugin.models``, the same way they contribute +sources and sinks: that is the plugin extension point for refutation evidence. + +Expected to remain red until ``abstract.ranges``, ``Validator`` and ``AuthorizationGuard`` +exist and the refutation engine consumes them. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager +from coretrace_python.findings import Confidence +from coretrace_python.findings.refutation import RefutationAnalysis, Status, Verdict, Verdicts +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.ir.model import Call, Symbol +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import ( + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + Source, + TaintKind, +) + +try: + from coretrace_python.abstract import Interval, RangeAnalysis + from coretrace_python.taint import AuthorizationGuard, Validator +except ImportError as error: # pragma: no cover - red until the evidence lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_evidence() -> None: + if MISSING is not None: + pytest.fail(f"refutation evidence is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +OS_SYSTEM = Sink(SymbolId("python.os.system"), TaintKind.COMMAND) +STDIN = Source(SymbolId("python.builtins.input"), "stdin") +LOGIN = SymbolId("python.flask_login.current_user.is_authenticated") +LOGIN_REQUIRED = SymbolId("python.flask_login.login_required") + +PRELUDE = "import os\nimport re\nfrom flask_login import current_user, login_required\n\n" + + +def manager_for(body: str, *models: object) -> AnalysisManager: + module = build_hir(SourceManager().add_source("proof.py", PRELUDE + body)) + manager = AnalysisManager(module) + manager.register(*engine.ALL_ANALYSES) + registry = SecurityModelRegistry() + registry.register(STDIN, OS_SYSTEM, *models) # type: ignore[arg-type] + manager.provide(SecurityModelAnalysis, registry.freeze()) + return manager + + +def first_function(manager: AnalysisManager) -> nodes.Function: + return next(s for s in manager.module.body if isinstance(s, nodes.Function)) + + +def verdict(body: str, *models: object) -> Verdict: + manager = manager_for(body, *models) + result: Verdicts = manager.get(RefutationAnalysis, first_function(manager)) + (found,) = result.all() + return found + + +def ranges_at_sink(body: str): # type: ignore[no-untyped-def] + """The intervals in the block calling ``os.system``, keyed by argument position.""" + + manager = manager_for(body) + function = first_function(manager) + ssa = manager.get(SSAAnalysis, function) + facts = manager.get(RangeAnalysis, function) + defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result} + for block in ssa.blocks: + for instruction in block.instructions: + callee = defs.get(instruction.callee) if isinstance(instruction, Call) else None + if isinstance(callee, Symbol) and callee.symbol_id == OS_SYSTEM.symbol: + at = facts.at(block.id) + return [at.get(argument) for argument in instruction.arguments] + raise AssertionError("no call to os.system") + + +# --------------------------------------------------------------------------- ranges + + +def test_intervals_join_and_widen() -> None: + assert Interval(0, 5).join(Interval(3, 9)) == Interval(0, 9) + assert Interval(0, 5).widen(Interval(0, 6)) == Interval(0, float("inf")) + assert Interval(0, 5).widen(Interval(-1, 5)) == Interval(float("-inf"), 5) + assert Interval(0, 5).widen(Interval(1, 4)) == Interval(0, 5) + assert str(Interval(0, float("inf"))) == "[0, inf]" + + +def test_numeric_constants_and_arithmetic_have_ranges() -> None: + assert ranges_at_sink("def f():\n n = 2 + 3 * 4\n os.system(n, n - 20, -n)\n") == [ + Interval(14, 14), + Interval(-6, -6), + Interval(-14, -14), + ] + + +def test_conversions_and_lengths_are_numeric() -> None: + low, length, raw = ranges_at_sink( + "def f(x):\n os.system(int(x), len(x), x)\n" + ) + assert low == Interval(float("-inf"), float("inf")) + assert length == Interval(0, float("inf")) + assert raw is None + + +def test_branches_refine_ranges_on_each_side() -> None: + then_side = ranges_at_sink("def f(x):\n n = int(x)\n if n < 10:\n os.system(n)\n") + else_side = ranges_at_sink("def f(x):\n n = int(x)\n if n < 10:\n pass\n else:\n os.system(n)\n") + chained = ranges_at_sink("def f(x):\n n = int(x)\n if 0 < n <= 100:\n os.system(n)\n") + + assert then_side == [Interval(float("-inf"), 10)] + assert else_side == [Interval(10, float("inf"))] + assert chained == [Interval(0, 100)] + + +def test_loops_terminate_by_widening() -> None: + (counter,) = ranges_at_sink( + "def f(x):\n i = 0\n while i < len(x):\n i = i + 1\n os.system(i)\n" + ) + assert counter == Interval(0, float("inf")) + + +def test_range_analysis_is_a_function_analysis_over_ssa() -> None: + from coretrace_python.cfg import CFGAnalysis + + assert RangeAnalysis.name == "abstract.ranges" + assert {SSAAnalysis, CFGAnalysis} <= RangeAnalysis.requires + assert RangeAnalysis in engine.ALL_ANALYSES + + +# --------------------------------------------------------------------------- numeric evidence + + +def test_numeric_values_cannot_inject() -> None: + found = verdict("def f():\n n = int(input())\n os.system('sleep ' + str(n))\n") + + assert found.status is Status.REFUTED + assert "numeric" in found.evidence + + +def test_ranges_show_in_the_evidence() -> None: + found = verdict( + "def f():\n n = int(input())\n if 0 <= n < 60:\n os.system('sleep ' + str(n))\n" + ) + assert found.status is Status.REFUTED + assert "[0, 60]" in found.evidence + + +def test_a_numeric_path_does_not_cover_a_raw_one() -> None: + found = verdict("def f():\n x = input()\n os.system(x + str(int(x)))\n") + assert found.status is Status.VULNERABILITY + + +def test_a_bounded_length_proves_nothing_about_the_string() -> None: + found = verdict("def f():\n x = input()\n if len(x) < 5:\n os.system(x)\n") + assert found.status is Status.HOTSPOT + + +# --------------------------------------------------------------------------- validators + + +FULLMATCH = Validator(SymbolId("python.re.fullmatch"), argument=1) if MISSING is None else None + + +def test_validators_and_authorization_guards_are_indexed_in_the_model_table() -> None: + guard = AuthorizationGuard(LOGIN, "login") + registry = SecurityModelRegistry() + registry.register(FULLMATCH, guard, Sink(LOGIN, TaintKind.HTML)) + table = registry.freeze() + + assert table.validator(FULLMATCH.symbol) == FULLMATCH + assert table.validator(LOGIN) is None + assert table.authorization(LOGIN) == guard + assert table.authorization(FULLMATCH.symbol) is None + assert table.validators == (FULLMATCH,) + assert table.authorizations == (guard,) + assert Validator(LOGIN).argument == 0 + extended = table.extended(Sink(SymbolId("python.x"), TaintKind.ADVISORY)) + assert extended.validators == (FULLMATCH,) and extended.authorizations == (guard,) + + +def test_a_true_validator_refutes_the_argument_it_validates() -> None: + found = verdict( + "def f():\n x = input()\n if re.fullmatch('[a-z]+', x):\n os.system(x)\n", + FULLMATCH, + ) + assert found.status is Status.REFUTED + assert found.evidence == "validated by python.re.fullmatch" + + +def test_a_validator_only_proves_its_declared_argument() -> None: + found = verdict( + "def f():\n x = input()\n if re.fullmatch(x, 'abc'):\n os.system(x)\n", + FULLMATCH, + ) + assert found.status is Status.HOTSPOT + + +def test_validators_on_the_wrong_branch_prove_nothing() -> None: + found = verdict( + "def f():\n x = input()\n if not re.fullmatch('[a-z]+', x):\n os.system(x)\n", + FULLMATCH, + ) + assert found.status is Status.VULNERABILITY + + +def test_compiled_pattern_validators_resolve_through_derived_symbols() -> None: + found = verdict( + "def f():\n x = input()\n pattern = re.compile('[a-z]+')\n" + " if pattern.fullmatch(x):\n os.system(x)\n", + Validator(SymbolId("python.re.compile.fullmatch")), + ) + assert found.status is Status.REFUTED + + +# --------------------------------------------------------------------------- authorization + + +def test_a_dominating_authorization_condition_makes_a_hotspot() -> None: + found = verdict( + "def f():\n if current_user.is_authenticated:\n os.system(input())\n", + AuthorizationGuard(LOGIN, "login"), + ) + assert found.status is Status.HOTSPOT + assert found.evidence == "behind authorization (login) at line 6" + + +def test_an_early_return_on_missing_authorization_counts() -> None: + found = verdict( + "def f():\n if not current_user.is_authenticated:\n return\n os.system(input())\n", + AuthorizationGuard(LOGIN, "login"), + ) + assert found.status is Status.HOTSPOT + + +def test_an_authorization_decorator_counts() -> None: + found = verdict( + "@login_required\ndef f():\n os.system(input())\n", + AuthorizationGuard(LOGIN_REQUIRED, "login"), + ) + assert found.status is Status.HOTSPOT + assert found.evidence == "behind authorization (login) by decorator" + + +def test_without_the_model_the_same_code_is_a_vulnerability() -> None: + found = verdict("def f():\n if current_user.is_authenticated:\n os.system(input())\n") + assert found.status is Status.VULNERABILITY + + +def test_authorization_does_not_override_a_refutation() -> None: + found = verdict( + "@login_required\ndef f():\n x = input()\n if x.isdigit():\n os.system(x)\n", + AuthorizationGuard(LOGIN_REQUIRED, "login"), + ) + assert found.status is Status.REFUTED + + +# --------------------------------------------------------------------------- plugins + + +def test_any_plugin_contributes_refutation_models() -> None: + class Evidence(ModelPlugin): + name = "evidence" + models = (FULLMATCH, AuthorizationGuard(LOGIN, "login")) + + table = engine.plugin_models([Evidence()]) + + assert table.validators == (FULLMATCH,) + assert table.authorizations == (AuthorizationGuard(LOGIN, "login"),) + + +def test_shipped_models_refute_fullmatch_and_flag_login_required_routes() -> None: + findings = engine.check( + SourceManager().add_source( + "app.py", + "import os\nimport re\nfrom flask import Flask, request\nfrom flask_login import login_required\n\n" + "app = Flask(__name__)\n\n" + "@app.route('/ping')\n" + "def ping():\n" + " host = request.args['host']\n" + " if re.fullmatch(r'[a-z.]+', host):\n" + " os.system('ping ' + host)\n\n" + "@app.route('/admin')\n" + "@login_required\n" + "def admin():\n" + " os.system(request.args['cmd'])\n", + ), + [PLUGINS], + ) + + assert [(f.function, f.confidence) for f in findings] == [("admin", Confidence.MEDIUM)] + assert findings[0].metadata["evidence"] == "behind authorization (login) by decorator" + + +def test_shipped_django_login_required_view_is_a_hotspot() -> None: + findings = engine.check( + SourceManager().add_source( + "views.py", + "import os\nfrom django.contrib.auth.decorators import login_required\n\n" + "@login_required\n" + "def run(request):\n" + " os.system(request.GET['cmd'])\n", + ), + [PLUGINS], + ) + assert [(f.rule_id, f.confidence) for f in findings] == [("command-injection", Confidence.MEDIUM)]