From 36d9d337b3e957755349ac736fef7837c7c33e5f Mon Sep 17 00:00:00 2001 From: Giathi-Daniel Date: Wed, 14 Oct 2026 15:54:53 +0300 Subject: [PATCH 1/2] Make build-dir skipping conditional, add scan_dev_files, parse IAM policies Spring CSRF skips stateless apps; Terraform policies are read structurally. --- CHANGELOG.md | 3 + devshield/core/config.py | 3 + devshield/core/context.py | 32 +++++- devshield/core/parsers/__init__.py | 4 +- devshield/core/parsers/hcl.py | 158 +++++++++++++++++++++++++- devshield/core/runner.py | 7 +- devshield/reports/rule_docs.py | 7 +- devshield/rules/_bases.py | 8 +- devshield/rules/_frameworks.py | 3 +- devshield/rules/cloud.py | 49 +++++--- devshield/rules/java.py | 3 +- devshield/rules/python.py | 14 ++- docs/configuration.md | 3 + docs/frameworks.md | 7 +- tests/core/test_framework_support.py | 48 +++++++- tests/rules/cases_frameworks.py | 11 ++ tests/rules/test_framework_details.py | 111 ++++++++++++++++++ tests/snapshots/spring-insecure.json | 2 +- 18 files changed, 428 insertions(+), 45 deletions(-) create mode 100644 tests/rules/test_framework_details.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b51b024..19b8ede 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,9 @@ All notable changes to this project are documented here. The format follows SH401-SH405), fixtures and docs. SH001-SH003 now cover these stacks too, and `devshield fix` creates starter CI workflows for Python, Java and .NET projects. +- `scan_dev_files` option to also check development-profile files, which the framework rules skip by + default. + ### Changed - Scans of large repositories are much faster: files that cannot match a rule are ruled out with diff --git a/devshield/core/config.py b/devshield/core/config.py index 0cb03f0..48c87f4 100644 --- a/devshield/core/config.py +++ b/devshield/core/config.py @@ -19,6 +19,8 @@ minimum: info fail_on: high exclude: [] +# Also check development-profile files such as application-dev.properties and settings/dev.py. +scan_dev_files: false report: format: terminal deps: @@ -78,6 +80,7 @@ class Config(_Section): ignore: list[str] = [] baseline: str | None = None exclude: list[str] = [] + scan_dev_files: bool = False severity: SeverityConfig = SeverityConfig() report: ReportConfig = ReportConfig() auto_fix: AutoFixConfig = AutoFixConfig() diff --git a/devshield/core/context.py b/devshield/core/context.py index 0940a52..6c1195b 100644 --- a/devshield/core/context.py +++ b/devshield/core/context.py @@ -1,6 +1,7 @@ from __future__ import annotations import ast +import fnmatch import json import os import re @@ -52,12 +53,18 @@ "coverage", "dist", "node_modules", - "obj", - "target", "vendor", "venv", } ) +# Directories that are build output only when a project file next to them says so, so that a +# folder that merely happens to be called "target" or "bin" is still scanned. +BUILD_OUTPUT = { + "target": ("pom.xml", "build.gradle", "build.gradle.kts", "Cargo.toml"), + "obj": ("*.csproj", "*.fsproj", "*.vbproj", "*.sln"), + "bin": ("*.csproj", "*.fsproj", "*.vbproj", "*.sln"), +} +DEV_FILE = re.compile(r"(?:^|[-_./])(?:dev|development|local|test|testing|debug)(?:[-_./]|$)", re.I) MAX_FILE_BYTES = 1_000_000 DOCKERFILE_PATTERNS = ("Dockerfile", "Dockerfile.*", "*.Dockerfile") COMPOSE_PATTERNS = ("docker-compose*.yml", "docker-compose*.yaml", "compose.yml", "compose.yaml") @@ -85,6 +92,11 @@ def _is_example(name: str) -> bool: return name.endswith(EXAMPLE_SUFFIXES) +def is_dev_path(rel: str) -> bool: + """Files meant for local development, where debug-style settings are expected.""" + return DEV_FILE.search(rel.rsplit("/", 1)[-1]) is not None + + def is_test_path(rel: str) -> bool: *directories, name = rel.split("/") return ( @@ -105,9 +117,11 @@ def __init__( exclude: Sequence[str] = (), respect_gitignore: bool = True, scan_history: bool = False, + scan_dev_files: bool = False, ) -> None: self.root = Path(root).resolve() self.scan_history = scan_history + self.scan_dev_files = scan_dev_files self._exclude = build_spec(exclude) self._respect_gitignore = respect_gitignore self._text: dict[str, str | None] = {} @@ -128,7 +142,7 @@ def files(self) -> tuple[str, ...]: for current, dirs, names in os.walk(self.root): parent = Path(current).relative_to(self.root).as_posix() parent = "" if parent == "." else parent - dirs[:] = sorted(d for d in dirs if not self._skip_dir(_join(parent, d))) + dirs[:] = sorted(d for d in dirs if not self._skip_dir(_join(parent, d), names)) for name in sorted(names): rel = _join(parent, name) if not self._exclude.match_file(rel) and ( @@ -137,9 +151,13 @@ def files(self) -> tuple[str, ...]: found.append(rel) return tuple(sorted(found)) - def _skip_dir(self, rel: str) -> bool: + def _skip_dir(self, rel: str, siblings: Sequence[str] = ()) -> bool: + name = rel.rsplit("/", 1)[-1] + markers = BUILD_OUTPUT.get(name, ()) + if any(fnmatch.fnmatch(s, marker) for s in siblings for marker in markers): + return True return ( - rel.rsplit("/", 1)[-1] in DEFAULT_EXCLUDES + name in DEFAULT_EXCLUDES or self._exclude.match_file(f"{rel}/") or self._gitignore.match_file(f"{rel}/") ) @@ -148,6 +166,10 @@ def match(self, *patterns: str) -> list[str]: spec = build_spec(patterns) return [f for f in self.files if spec.match_file(f)] + def is_dev_file(self, rel: str) -> bool: + """True for development-profile files, unless ``scan_dev_files`` turns the skip off.""" + return not self.scan_dev_files and is_dev_path(rel) + def is_gitignored(self, rel: str) -> bool: if rel not in self._ignored: self._ignored[rel] = self._gitignore.match_file(rel) diff --git a/devshield/core/parsers/__init__.py b/devshield/core/parsers/__init__.py index 464dfcc..2de1b5f 100644 --- a/devshield/core/parsers/__init__.py +++ b/devshield/core/parsers/__init__.py @@ -14,7 +14,7 @@ stages, ) from devshield.core.parsers.dotenv import EnvEntry, parse_dotenv -from devshield.core.parsers.hcl import HclBlock, parse_hcl +from devshield.core.parsers.hcl import HclBlock, Opaque, parse_hcl, parse_hcl_value from devshield.core.parsers.php import PhpEnv, PhpExpr, parse_php_config from devshield.core.parsers.pyast import ( MISSING, @@ -38,6 +38,7 @@ "EnvEntry", "HclBlock", "Instruction", + "Opaque", "PhpEnv", "PhpExpr", "effective_user", @@ -53,6 +54,7 @@ "parse_dockerfile", "parse_dotenv", "parse_hcl", + "parse_hcl_value", "parse_module", "parse_php_config", "parse_pipfile", diff --git a/devshield/core/parsers/hcl.py b/devshield/core/parsers/hcl.py index 9aa678d..1456628 100644 --- a/devshield/core/parsers/hcl.py +++ b/devshield/core/parsers/hcl.py @@ -6,6 +6,7 @@ from __future__ import annotations +import json import re from dataclasses import dataclass, field @@ -127,4 +128,159 @@ def depth(text: str) -> int: parts.append(part) total += depth(part) index += 1 - return " ".join(parts), index + return "\n".join(parts), index + + +@dataclass(frozen=True) +class Opaque: + """An expression that is not a literal, such as ``aws_s3_bucket.b.arn`` or ``var.name``.""" + + text: str + + +_TOKEN = re.compile( + r"""\s*(?:(?P\n)|(?P"(?:[^"\\]|\\.)*")|(?P-?\d+(?:\.\d+)?)""" + r"""|(?P[A-Za-z_][\w\-]*(?:\.[\w\-*]+|\[\d+\]|\[\*\])*)|(?P=>|[{}\[\](),=:?]))""" +) + + +def _tokens(text: str) -> list[tuple[str, str]]: + found: list[tuple[str, str]] = [] + position = 0 + while position < len(text): + stripped = text[position:].lstrip(" \t\r") + position = len(text) - len(stripped) + match = _TOKEN.match(text, position) + if not match or match.end() == position: + position += 1 # an operator or character the reader does not need + continue + position = match.end() + kind = match.lastgroup or "" + found.append((kind, match.group(kind))) + return found + + +class _ValueParser: + def __init__(self, tokens: list[tuple[str, str]]) -> None: + self.tokens = tokens + self.position = 0 + + def _peek(self) -> tuple[str, str]: + return self.tokens[self.position] if self.position < len(self.tokens) else ("end", "") + + def _skip_newlines(self) -> None: + while self._peek()[0] == "nl": + self.position += 1 + + def value(self) -> object: + self._skip_newlines() + kind, text = self._peek() + if kind == "string": + self.position += 1 + return text[1:-1] + if kind == "number": + self.position += 1 + return float(text) if "." in text else int(text) + if text == "{": + return self._object() + if text == "[": + return self._list() + if kind == "word": + self.position += 1 + return self._word(text) + return self._opaque() + + def _word(self, word: str) -> object: + if word in {"true", "false"}: + return word == "true" + if word == "null": + return None + if self._peek()[1] == "(": # a function call: jsonencode(x) is x, anything else is opaque + self.position += 1 + arguments = self._arguments() + return arguments[0] if word == "jsonencode" and arguments else Opaque(word + "(...)") + if self._peek()[1] == "?": # a conditional expression: cond ? a : b + return Opaque(f"{word} {self._opaque().text}") + return Opaque(word) + + def _arguments(self) -> list[object]: + found: list[object] = [] + while True: + self._skip_newlines() + kind, text = self._peek() + if kind == "end" or text == ")": + self.position += 1 + return found + if text == ",": + self.position += 1 + continue + found.append(self.value()) + + def _object(self) -> dict[str, object]: + self.position += 1 + result: dict[str, object] = {} + while True: + self._skip_newlines() + kind, text = self._peek() + if kind == "end": + return result + if text == "}": + self.position += 1 + return result + if text == ",": + self.position += 1 + continue + if kind not in {"word", "string"}: + self.position += 1 + continue + self.position += 1 + key = text[1:-1] if kind == "string" else text + self._skip_newlines() + if self._peek()[1] in {"=", ":"}: + self.position += 1 + result[key] = self.value() + + def _list(self) -> list[object]: + self.position += 1 + result: list[object] = [] + while True: + self._skip_newlines() + kind, text = self._peek() + if kind == "end": + return result + if text == "]": + self.position += 1 + return result + if text == ",": + self.position += 1 + continue + result.append(self.value()) + + def _opaque(self) -> Opaque: + """Skip one unreadable expression up to the next separator at the same depth.""" + start, depth = self.position, 0 + while self.position < len(self.tokens): + kind, text = self.tokens[self.position] + if depth == 0 and (kind == "nl" or text in {",", "}", "]", ")"}): + break + depth += text in {"{", "[", "("} + depth -= text in {"}", "]", ")"} + self.position += 1 + if self.position == start: + self.position += 1 + return Opaque(" ".join(t for _, t in self.tokens[start : self.position])) + + +def parse_hcl_value(text: str) -> object: + """Read an HCL literal (object, list, string, number, boolean), unwrapping ``jsonencode``. + + Anything that is not a literal comes back as ``Opaque``. Heredoc JSON is decoded as JSON. + """ + stripped = text.strip() + heredoc = re.match(r"<<-?\s*(\w+)\n(.*)\n\s*\1\s*$", stripped, re.S) + if heredoc: + try: + return json.loads(heredoc.group(2)) + except ValueError: + return Opaque("heredoc") + return _ValueParser(_tokens(stripped)).value() diff --git a/devshield/core/runner.py b/devshield/core/runner.py index 6fb363f..acb0452 100644 --- a/devshield/core/runner.py +++ b/devshield/core/runner.py @@ -70,7 +70,12 @@ def run_scan( config = config or load_config(root) registry = registry or default_registry() options = options or ScanOptions() - ctx = ProjectContext(root, exclude=config.exclude, scan_history=options.history) + ctx = ProjectContext( + root, + exclude=config.exclude, + scan_history=options.history, + scan_dev_files=config.scan_dev_files, + ) rules_run: list[str] = [] timings: dict[str, float] = {} diff --git a/devshield/reports/rule_docs.py b/devshield/reports/rule_docs.py index c7a4f2f..b8dcdc2 100644 --- a/devshield/reports/rule_docs.py +++ b/devshield/reports/rule_docs.py @@ -100,8 +100,11 @@ def render_all(rules: Iterable[type[Rule]]) -> dict[str, str]: | CloudFormation | YAML or JSON templates that declare `AWS::` resources | | Docker, GitHub Actions | `Dockerfile` or compose files, workflow files | -Detected stacks are listed at the top of every report. Build output (`target/`, `obj/`, -`.terraform/`) and virtual environments are never scanned. +Detected stacks are listed at the top of every report. Build output is skipped: `target/` next to a +`pom.xml` or Gradle build, `obj/` and `bin/` next to a `.csproj`, `.terraform/`, and virtual +environments. A folder that only shares one of those names is still scanned. Development-profile +files (`application-dev.properties`, `settings/dev.py`) are skipped by the debug-style checks; set +`scan_dev_files: true` in `devshield.yml` to check them too. ## Rules by stack diff --git a/devshield/rules/_bases.py b/devshield/rules/_bases.py index f445a5b..3d81787 100644 --- a/devshield/rules/_bases.py +++ b/devshield/rules/_bases.py @@ -4,6 +4,7 @@ from collections.abc import Iterable, Iterator from typing import TYPE_CHECKING, ClassVar +from devshield.core.context import is_dev_path from devshield.core.models import Finding, Framework from devshield.core.rule import Rule from devshield.core.search import Hit, search @@ -31,12 +32,7 @@ def is_deployed_env(ctx: ProjectContext, file: str, values: dict[str, str]) -> b ) -DEV_FILE = re.compile(r"(?:^|[-_./])(?:dev|development|local|test|testing|debug)(?:[-_./]|$)", re.I) - - -def is_dev_file(rel: str) -> bool: - """Files meant for local development, where debug-style settings are expected.""" - return DEV_FILE.search(rel.rsplit("/", 1)[-1]) is not None +is_dev_file = is_dev_path def is_application(ctx: ProjectContext) -> bool: diff --git a/devshield/rules/_frameworks.py b/devshield/rules/_frameworks.py index 820bb33..134340f 100644 --- a/devshield/rules/_frameworks.py +++ b/devshield/rules/_frameworks.py @@ -6,7 +6,6 @@ from devshield.core.context import ProjectContext from devshield.core.parsers import module_assignments -from devshield.rules._bases import is_dev_file def django_settings(ctx: ProjectContext, include_dev: bool = False) -> list[str]: @@ -16,7 +15,7 @@ def django_settings(ctx: ProjectContext, include_dev: bool = False) -> list[str] for f in ctx.python_files if f.rsplit("/", 1)[-1] == "settings.py" or "/settings/" in f"/{f}" ] - return files if include_dev else [f for f in files if not is_dev_file(f)] + return files if include_dev else [f for f in files if not ctx.is_dev_file(f)] def settings_values( diff --git a/devshield/rules/cloud.py b/devshield/rules/cloud.py index 650321c..19f00e4 100644 --- a/devshield/rules/cloud.py +++ b/devshield/rules/cloud.py @@ -9,7 +9,7 @@ from devshield.core.context import ProjectContext from devshield.core.models import Category, Finding, Framework, Severity -from devshield.core.parsers import HclBlock +from devshield.core.parsers import HclBlock, parse_hcl_value from devshield.core.rule import Rule IAC = frozenset({Framework.TERRAFORM, Framework.CLOUDFORMATION}) @@ -95,6 +95,20 @@ def cfn_statements(document: Any) -> Iterator[dict[str, Any]]: yield statement +def hcl_statements(text: str) -> list[dict[str, Any]] | None: + """Statements of an inline Terraform policy (jsonencode or heredoc JSON), else None.""" + data = parse_hcl_value(text) + statements = data.get("Statement") if isinstance(data, dict) else None + if statements is None: + return None + return [s for s in _as_list(statements) if isinstance(s, dict)] + + +def _everyone(statement: dict[str, Any]) -> bool: + principal = statement.get("Principal") + return principal == "*" or (isinstance(principal, dict) and _has_star(principal.get("AWS"))) + + def _has_star(value: Any) -> bool: return any(item == "*" for item in _as_list(value)) @@ -136,7 +150,7 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: file=file, line=block.line_of(setting), ) - elif kind == "aws_s3_bucket_policy" and self._public_policy( + elif kind == "aws_s3_bucket_policy" and self._public_hcl_policy( block.value("policy") or "" ): yield self.finding( @@ -168,13 +182,9 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: ) elif kind == "AWS::S3::BucketPolicy": for statement in cfn_statements(props.get("PolicyDocument")): - principal = statement.get("Principal") - everyone = principal == "*" or ( - isinstance(principal, dict) and _has_star(principal.get("AWS")) - ) if ( statement.get("Effect") == "Allow" - and everyone + and _everyone(statement) and "Condition" not in statement ): yield self.finding( @@ -183,6 +193,15 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: line=cfn_line(ctx, file, logical_id, "PolicyDocument"), ) + @classmethod + def _public_hcl_policy(cls, text: str) -> bool: + statements = hcl_statements(text) + if statements is None: + return cls._public_policy(text) + return any( + s.get("Effect") == "Allow" and _everyone(s) and "Condition" not in s for s in statements + ) + @staticmethod def _public_policy(text: str) -> bool: return ( @@ -209,12 +228,16 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: for file, block in terraform_blocks(ctx): if block.kind == "resource" and block.type in IAM_POLICY_TYPES: text = block.value("policy") or "" - if "Deny" in text: - continue - if POLICY_ACTION.search(text): - yield self._wildcard( - file, block.line_of("policy"), bool(POLICY_RESOURCE.search(text)) - ) + line = block.line_of("policy") + statements = hcl_statements(text) + if statements is not None: + for statement in statements: + if statement.get("Effect") == "Allow" and _has_star( + statement.get("Action") + ): + yield self._wildcard(file, line, _has_star(statement.get("Resource"))) + elif "Deny" not in text and POLICY_ACTION.search(text): + yield self._wildcard(file, line, bool(POLICY_RESOURCE.search(text))) elif block.kind == "data" and block.type == "aws_iam_policy_document": for item in block.children("statement"): effect = unquote(item.value("effect")) or "Allow" diff --git a/devshield/rules/java.py b/devshield/rules/java.py index a355dd8..b98fc65 100644 --- a/devshield/rules/java.py +++ b/devshield/rules/java.py @@ -7,7 +7,6 @@ from devshield.core.models import Category, Finding, Framework, Severity from devshield.core.rule import Rule from devshield.core.search import search_text -from devshield.rules._bases import is_dev_file from devshield.rules._frameworks import spring_config_files SPRING = frozenset({Framework.SPRING}) @@ -23,7 +22,7 @@ def _entries(ctx: ProjectContext) -> Iterator[tuple[str, str, str, int]]: """(file, lowercase key, value, line) of every production Spring config entry.""" for file in spring_config_files(ctx): - if is_dev_file(file): + if ctx.is_dev_file(file): continue for entry in ctx.config_entries(file): yield file, entry.key.lower().replace("_", "-"), entry.value.strip(), entry.line diff --git a/devshield/rules/python.py b/devshield/rules/python.py index 1c1fd7f..cf772d7 100644 --- a/devshield/rules/python.py +++ b/devshield/rules/python.py @@ -10,7 +10,6 @@ from devshield.core.parsers import string_literals as string_values from devshield.core.rule import Rule from devshield.core.search import call_arguments, is_string_literal, search, search_text -from devshield.rules._bases import is_dev_file from devshield.rules._frameworks import ( DOTNET_COOKIE, SPRING_COOKIE_KEYS, @@ -130,7 +129,7 @@ def _django(self, ctx: ProjectContext) -> Iterator[Finding]: def _spring(self, ctx: ProjectContext) -> Iterator[Finding]: for file in spring_config_files(ctx): - if is_dev_file(file): + if ctx.is_dev_file(file): continue for entry in ctx.config_entries(file): key = entry.key.lower() @@ -327,6 +326,9 @@ class CsrfDisabled(Rule): r"\bcsrf\s*\(\s*\)\s*\.\s*disable\s*\(|\bcsrf\s*\(\s*(?:\w+\s*->\s*\w+\s*\.\s*disable\s*\(" r"|[\w.]*::\s*disable)" ) + stateless = re.compile( + r"SessionCreationPolicy\.STATELESS|oauth2ResourceServer|jwtAuthenticationConverter" + ) dotnet = re.compile(r"\[IgnoreAntiforgeryToken\]|IgnoreAntiforgeryTokenAttribute") def check(self, ctx: ProjectContext) -> Iterator[Finding]: @@ -341,8 +343,12 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: severity=Severity.LOW, ) for hit in search_text(ctx, ctx.java_files, self.spring): + if self.stateless.search(ctx.read_text(hit.file) or ""): + continue # token-based APIs do not need CSRF protection yield self.finding( - "Spring Security CSRF protection is disabled", file=hit.file, line=hit.line + "CSRF protection is disabled and the app is not configured as stateless", + file=hit.file, + line=hit.line, ) for hit in search(ctx, ctx.csharp_files, self.dotnet): yield self.finding("Antiforgery validation is skipped", file=hit.file, line=hit.line) @@ -383,7 +389,7 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: if "secret_key" not in text.lower(): continue tree = ctx.pyast(file) - if tree is None or is_dev_file(file): + if tree is None or ctx.is_dev_file(file): continue for name, value, line in secret_assignments(tree, PYTHON_SECRET_NAMES): literals = [s for s in string_values(value) if s.strip()] diff --git a/docs/configuration.md b/docs/configuration.md index 5372747..8e8dc21 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -14,6 +14,9 @@ ignore: exclude: - tests/fixtures/ +# Also check development-profile files (application-dev.properties, settings/dev.py, ...). +scan_dev_files: false + severity: minimum: info # hide findings below this: info, low, medium, high, critical fail_on: high # exit 1 at this severity or above diff --git a/docs/frameworks.md b/docs/frameworks.md index fec8255..f758089 100644 --- a/docs/frameworks.md +++ b/docs/frameworks.md @@ -18,8 +18,11 @@ map from stack to rules; each ID links to its own page. | CloudFormation | YAML or JSON templates that declare `AWS::` resources | | Docker, GitHub Actions | `Dockerfile` or compose files, workflow files | -Detected stacks are listed at the top of every report. Build output (`target/`, `obj/`, -`.terraform/`) and virtual environments are never scanned. +Detected stacks are listed at the top of every report. Build output is skipped: `target/` next to a +`pom.xml` or Gradle build, `obj/` and `bin/` next to a `.csproj`, `.terraform/`, and virtual +environments. A folder that only shares one of those names is still scanned. Development-profile +files (`application-dev.properties`, `settings/dev.py`) are skipped by the debug-style checks; set +`scan_dev_files: true` in `devshield.yml` to check them too. ## Rules by stack diff --git a/tests/core/test_framework_support.py b/tests/core/test_framework_support.py index 566690a..65b7851 100644 --- a/tests/core/test_framework_support.py +++ b/tests/core/test_framework_support.py @@ -100,15 +100,28 @@ def test_workflow_files_are_not_cloudformation(self, make_project: MakeProject) ) assert Framework.CLOUDFORMATION not in detected(root) - def test_build_output_directories_are_skipped(self, make_project: MakeProject) -> None: + def test_build_output_is_skipped_only_next_to_its_project_file( + self, make_project: MakeProject + ) -> None: root = make_project( { + "pom.xml": "", "target/classes/application.properties": "x=1\n", - ".terraform/modules/m/main.tf": 'resource "a" "b" {}\n', + "App.csproj": "", "obj/Debug/Gen.cs": "class A {}\n", + "bin/Debug/App.cs": "class B {}\n", + ".terraform/modules/m/main.tf": 'resource "a" "b" {}\n', } ) - assert ProjectContext(root).files == () + assert ProjectContext(root).files == ("App.csproj", "pom.xml") + + def test_folders_that_only_share_a_name_are_still_scanned( + self, make_project: MakeProject + ) -> None: + root = make_project( + {"app.py": "x = 1\n", "target/goals.py": "y = 2\n", "bin/run.sh": "echo\n"} + ) + assert ProjectContext(root).files == ("app.py", "bin/run.sh", "target/goals.py") def test_test_paths_for_the_new_ecosystems() -> None: @@ -187,10 +200,10 @@ def test_hcl_blocks_attributes_and_nested_blocks(self) -> None: first, second, third = parse_hcl(text) assert (first.kind, first.labels, first.line) == ("resource", ("aws_s3_bucket", "b"), 2) assert first.value("bucket") == '"x"' and first.line_of("bucket") == 3 - assert first.value("tags") == '{ a = "b" }' + assert first.value("tags") == '{\na = "b"\n}' assert first.children("versioning")[0].value("enabled") == "true" assert '{"a": "}"}' in (first.value("policy") or "") - assert first.value("list") == '[ "one", "two", ]' + assert first.value("list") == '[\n"one",\n"two",\n]' assert second.value("storage_encrypted") == "false" assert third.kind == "data" and third.children("statement")[0].value("effect") == '"Allow"' assert parse_hcl("nothing here") == [] @@ -217,3 +230,28 @@ def test_python_ast_helpers(self) -> None: literals = {line: string_literals(value) for _, value, line in found} assert literals == {5: ["fallback"], 6: ["dev"], 7: ["z"], 8: []} assert parse_module("def (") is None + + +class TestHclValues: + def test_literals_lists_objects_and_jsonencode(self) -> None: + from devshield.core.parsers import Opaque, parse_hcl_value + + value = parse_hcl_value( + 'jsonencode({\n a = "x"\n "b" : [1, 2.5, true, null]\n c = { d = var.y }\n' + " e = merge(local.a, local.b)\n f = cond ? 1 : 2\n})" + ) + assert value == { + "a": "x", + "b": [1, 2.5, True, None], + "c": {"d": Opaque("var.y")}, + "e": Opaque("merge(...)"), + "f": Opaque("cond ? 1 : 2"), + } + + def test_heredoc_json_and_garbage(self) -> None: + from devshield.core.parsers import Opaque, parse_hcl_value + + assert parse_hcl_value('<<-EOT\n {"a": [1]}\n EOT') == {"a": [1]} + assert parse_hcl_value("< c.disable())\n" + " .sessionManagement(m -> m.sessionCreationPolicy(SessionCreationPolicy.STATELESS));\n" + ), + }, + { + **SPRING, + "src/main/java/S.java": "http.csrf().disable().oauth2ResourceServer(o -> o.jwt());\n", + }, ], "SH108": [ { diff --git a/tests/rules/test_framework_details.py b/tests/rules/test_framework_details.py new file mode 100644 index 0000000..2637e59 --- /dev/null +++ b/tests/rules/test_framework_details.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +from typing import ClassVar + +import pytest +from devshield.core.config import Config +from devshield.core.models import Severity +from devshield.core.runner import ScanOptions, run_scan + +from tests.conftest import MakeProject +from tests.rules.helpers import REGISTRY, SPRING + + +def scan(root: object, *rules: str, config: Config | None = None): # type: ignore[no-untyped-def] + return run_scan(root, config or Config(), REGISTRY, ScanOptions(only=rules)) # type: ignore[arg-type] + + +def policy(statements: str) -> dict[str, str]: + return { + "main.tf": ( + 'resource "aws_iam_policy" "p" {\n policy = jsonencode({\n Version = "2012-10-17"\n' + f" Statement = [\n{statements}\n ]\n }})\n}}\n" + ) + } + + +class TestIamPolicyParsing: + def test_statements_are_judged_one_by_one(self, make_project: MakeProject) -> None: + # Two statements that each hold one wildcard must not be merged into "admin access". + root = make_project( + policy( + ' { Effect = "Allow", Action = "*", Resource = "arn:aws:s3:::x" },\n' + ' { Effect = "Allow", Action = "s3:GetObject", Resource = "*" }' + ) + ) + [finding] = scan(root, "SH402").findings + assert finding.severity is Severity.HIGH and finding.message == "Allows every action" + + def test_admin_statement_is_critical(self, make_project: MakeProject) -> None: + root = make_project( + policy(' { Effect = "Allow", Action = ["s3:Get*", "*"], Resource = ["*"] }') + ) + [finding] = scan(root, "SH402").findings + assert finding.severity is Severity.CRITICAL + + def test_deny_statements_and_references_are_not_flagged( + self, make_project: MakeProject + ) -> None: + root = make_project( + policy( + ' { Effect = "Deny", Action = "*", Resource = "*" },\n' + ' {\n Effect = "Allow"\n Action = var.actions\n' + " Resource = aws_s3_bucket.b.arn\n }" + ) + ) + assert scan(root, "SH402").findings == () + + def test_heredoc_json_policies_are_read(self, make_project: MakeProject) -> None: + root = make_project( + { + "main.tf": ( + 'resource "aws_iam_policy" "p" {\n policy = < None: + def bucket(statement: str) -> dict[str, str]: + return { + "main.tf": ( + 'resource "aws_s3_bucket_policy" "p" {\n policy = jsonencode({\n' + f" Statement = [{statement}]\n }})\n}}\n" + ) + } + + public = '{ Effect = "Allow", Principal = { AWS = "*" }, Action = "s3:GetObject" }' + conditional = ( + '{ Effect = "Allow", Principal = "*", Action = "s3:GetObject", ' + 'Condition = { StringEquals = { "aws:SourceVpce" = "vpce-1" } } }' + ) + assert len(scan(make_project(bucket(public)), "SH401").findings) == 1 + assert scan(make_project(bucket(conditional)), "SH401").findings == () + + +class TestDevProfiles: + FILES: ClassVar[dict[str, str]] = { + **SPRING, + "src/main/resources/application-dev.properties": "management.endpoints.web.exposure.include=*\n", + } + + def test_dev_profile_files_are_skipped_by_default(self, make_project: MakeProject) -> None: + assert scan(make_project(self.FILES), "SH018").findings == () + + def test_scan_dev_files_turns_the_skip_off(self, make_project: MakeProject) -> None: + found = scan(make_project(self.FILES), "SH018", config=Config(scan_dev_files=True)) + assert [f.rule_id for f in found.findings] == ["SH018"] + + def test_the_option_reaches_the_python_rules(self, make_project: MakeProject) -> None: + files = {"requirements.txt": "Django\n", "web/settings/dev.py": "ALLOWED_HOSTS = ['*']\n"} + root = make_project(files) + assert scan(root, "SH014").findings == () + assert len(scan(root, "SH014", config=Config(scan_dev_files=True)).findings) == 1 + + @pytest.mark.parametrize("value", ["true", "false"]) + def test_config_file_accepts_the_option(self, make_project: MakeProject, value: str) -> None: + from devshield.core.config import load_config + + root = make_project({"devshield.yml": f"scan_dev_files: {value}\n"}) + assert load_config(root).scan_dev_files is (value == "true") diff --git a/tests/snapshots/spring-insecure.json b/tests/snapshots/spring-insecure.json index 3724254..4f826f4 100644 --- a/tests/snapshots/spring-insecure.json +++ b/tests/snapshots/spring-insecure.json @@ -119,7 +119,7 @@ "title": "CSRF protection disabled", "severity": "medium", "category": "authentication", - "message": "Spring Security CSRF protection is disabled", + "message": "CSRF protection is disabled and the app is not configured as stateless", "location": { "file": "src/main/java/com/example/SecurityConfig.java", "line": 12, From f3790f11065eda588e91a428f126c356c79bc557 Mon Sep 17 00:00:00 2001 From: Giathi-Daniel Date: Wed, 14 Oct 2026 15:58:26 +0300 Subject: [PATCH 2/2] Resolve Terraform locals, variable defaults, concat and merge in IAM and S3 rules Policies built from references are now evaluated per module directory. --- CHANGELOG.md | 3 ++ devshield/core/parsers/__init__.py | 11 +++- devshield/core/parsers/hcl.py | 76 +++++++++++++++++++++++++-- devshield/rules/cloud.py | 24 ++++++--- tests/core/test_framework_support.py | 12 ++++- tests/rules/test_framework_details.py | 73 +++++++++++++++++++++++++ 6 files changed, 185 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 19b8ede..b4bb92a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,9 @@ All notable changes to this project are documented here. The format follows - `scan_dev_files` option to also check development-profile files, which the framework rules skip by default. +- Terraform policies that are built from `locals`, variable defaults, `concat()` and `merge()` are + now evaluated per module, so the IAM and S3 rules see through them. + ### Changed - Scans of large repositories are much faster: files that cannot match a rule are ruled out with diff --git a/devshield/core/parsers/__init__.py b/devshield/core/parsers/__init__.py index 2de1b5f..37006dc 100644 --- a/devshield/core/parsers/__init__.py +++ b/devshield/core/parsers/__init__.py @@ -14,7 +14,14 @@ stages, ) from devshield.core.parsers.dotenv import EnvEntry, parse_dotenv -from devshield.core.parsers.hcl import HclBlock, Opaque, parse_hcl, parse_hcl_value +from devshield.core.parsers.hcl import ( + Call, + HclBlock, + Opaque, + TerraformScope, + parse_hcl, + parse_hcl_value, +) from devshield.core.parsers.php import PhpEnv, PhpExpr, parse_php_config from devshield.core.parsers.pyast import ( MISSING, @@ -34,6 +41,7 @@ __all__ = [ "MISSING", + "Call", "ConfigEntry", "EnvEntry", "HclBlock", @@ -41,6 +49,7 @@ "Opaque", "PhpEnv", "PhpExpr", + "TerraformScope", "effective_user", "final_stage", "flatten_json", diff --git a/devshield/core/parsers/hcl.py b/devshield/core/parsers/hcl.py index 1456628..40a4c01 100644 --- a/devshield/core/parsers/hcl.py +++ b/devshield/core/parsers/hcl.py @@ -8,9 +8,13 @@ import json import re +from collections.abc import Iterable from dataclasses import dataclass, field -_HEADER = re.compile(r'^\s*(resource|data|module|provider)\s+((?:"[^"]*"\s*)+|[\w-]+\s*)\{', re.M) +_HEADER = re.compile( + r'^\s*(resource|data|module|provider|variable|output|locals)(?:\s+((?:"[^"]*"\s*)+|[\w-]+\s*))?\s*\{', + re.M, +) _BLOCK = re.compile(r'^\s*([\w-]+)((?:\s+"[^"]*")*)\s*\{\s*$') _ATTRIBUTE = re.compile(r"^\s*([\w-]+)\s*=\s*(.*)$") _HEREDOC = re.compile(r"<<-?\s*([A-Za-z_][\w]*)\s*$") @@ -40,7 +44,9 @@ def children(self, name: str) -> list[HclBlock]: return [b for b in self.blocks if b.kind == name] -def _labels(text: str) -> tuple[str, ...]: +def _labels(text: str | None) -> tuple[str, ...]: + if not text: + return () return tuple(re.findall(r'"([^"]*)"', text)) or tuple(text.split()) @@ -55,7 +61,7 @@ def _strip_comment(line: str) -> str: def parse_hcl(text: str) -> list[HclBlock]: - """Top-level resource, data, module and provider blocks.""" + """Top-level resource, data, module, provider, variable, output and locals blocks.""" lines = text.splitlines() blocks: list[HclBlock] = [] index = 0 @@ -160,6 +166,14 @@ def _tokens(text: str) -> list[tuple[str, str]]: return found +@dataclass(frozen=True) +class Call: + """A function call such as ``concat(a, b)`` or ``merge(a, b)``.""" + + name: str + arguments: tuple[object, ...] + + class _ValueParser: def __init__(self, tokens: list[tuple[str, str]]) -> None: self.tokens = tokens @@ -198,7 +212,9 @@ def _word(self, word: str) -> object: if self._peek()[1] == "(": # a function call: jsonencode(x) is x, anything else is opaque self.position += 1 arguments = self._arguments() - return arguments[0] if word == "jsonencode" and arguments else Opaque(word + "(...)") + if word == "jsonencode" and arguments: + return arguments[0] + return Call(word, tuple(arguments)) if self._peek()[1] == "?": # a conditional expression: cond ? a : b return Opaque(f"{word} {self._opaque().text}") return Opaque(word) @@ -284,3 +300,55 @@ def parse_hcl_value(text: str) -> object: except ValueError: return Opaque("heredoc") return _ValueParser(_tokens(stripped)).value() + + +_REFERENCE = re.compile(r"(local|var)\.(\w+)") +MAX_RESOLVE_DEPTH = 8 + + +class TerraformScope: + """The locals and variable defaults of one Terraform module (one directory).""" + + def __init__(self, blocks: Iterable[HclBlock]) -> None: + self.locals: dict[str, str] = {} + self.variables: dict[str, str] = {} + for block in blocks: + if block.kind == "locals": + self.locals.update({name: raw for name, (raw, _) in block.attributes.items()}) + elif block.kind == "variable" and block.labels and block.value("default") is not None: + self.variables[block.labels[0]] = block.value("default") or "" + + def value(self, text: str) -> object: + """``parse_hcl_value`` with local.x, var.x, concat() and merge() evaluated.""" + return self.resolve(parse_hcl_value(text), 0) + + def resolve(self, value: object, depth: int) -> object: + if depth > MAX_RESOLVE_DEPTH: + return Opaque("too deep") + if isinstance(value, Opaque): + match = _REFERENCE.fullmatch(value.text) + raw = None + if match: + source = self.locals if match.group(1) == "local" else self.variables + raw = source.get(match.group(2)) + return self.resolve(parse_hcl_value(raw), depth + 1) if raw is not None else value + if isinstance(value, dict): + return {k: self.resolve(v, depth + 1) for k, v in value.items()} + if isinstance(value, list): + return [self.resolve(v, depth + 1) for v in value] + if isinstance(value, Call): + return self._call(value, depth) + return value + + def _call(self, call: Call, depth: int) -> object: + arguments = [self.resolve(a, depth + 1) for a in call.arguments] + lists = [a for a in arguments if isinstance(a, list)] + objects = [a for a in arguments if isinstance(a, dict)] + if call.name == "concat" and len(lists) == len(arguments): + return [item for items in lists for item in items] + if call.name == "merge" and len(objects) == len(arguments): + merged: dict[str, object] = {} + for mapping in objects: + merged.update(mapping) + return merged + return Opaque(f"{call.name}(...)") diff --git a/devshield/rules/cloud.py b/devshield/rules/cloud.py index 19f00e4..a0e2761 100644 --- a/devshield/rules/cloud.py +++ b/devshield/rules/cloud.py @@ -9,7 +9,7 @@ from devshield.core.context import ProjectContext from devshield.core.models import Category, Finding, Framework, Severity -from devshield.core.parsers import HclBlock, parse_hcl_value +from devshield.core.parsers import HclBlock, TerraformScope from devshield.core.rule import Rule IAC = frozenset({Framework.TERRAFORM, Framework.CLOUDFORMATION}) @@ -95,9 +95,17 @@ def cfn_statements(document: Any) -> Iterator[dict[str, Any]]: yield statement -def hcl_statements(text: str) -> list[dict[str, Any]] | None: +def terraform_scopes(ctx: ProjectContext) -> dict[str, TerraformScope]: + """One scope per directory: Terraform locals and variables belong to a module.""" + by_directory: dict[str, list[HclBlock]] = {} + for file in ctx.terraform_files: + by_directory.setdefault(file.rpartition("/")[0], []).extend(ctx.terraform(file)) + return {directory: TerraformScope(blocks) for directory, blocks in by_directory.items()} + + +def hcl_statements(text: str, scope: TerraformScope) -> list[dict[str, Any]] | None: """Statements of an inline Terraform policy (jsonencode or heredoc JSON), else None.""" - data = parse_hcl_value(text) + data = scope.value(text) statements = data.get("Statement") if isinstance(data, dict) else None if statements is None: return None @@ -129,6 +137,7 @@ class PublicBucket(Rule): ) def check(self, ctx: ProjectContext) -> Iterator[Finding]: + scopes = terraform_scopes(ctx) for file, block in terraform_blocks(ctx): if block.kind != "resource": continue @@ -151,7 +160,7 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: line=block.line_of(setting), ) elif kind == "aws_s3_bucket_policy" and self._public_hcl_policy( - block.value("policy") or "" + block.value("policy") or "", scopes[file.rpartition("/")[0]] ): yield self.finding( "Bucket policy allows every principal", @@ -194,8 +203,8 @@ def check(self, ctx: ProjectContext) -> Iterator[Finding]: ) @classmethod - def _public_hcl_policy(cls, text: str) -> bool: - statements = hcl_statements(text) + def _public_hcl_policy(cls, text: str, scope: TerraformScope) -> bool: + statements = hcl_statements(text, scope) if statements is None: return cls._public_policy(text) return any( @@ -225,11 +234,12 @@ class WildcardIamPolicy(Rule): remediation = "List the specific actions and resources the workload needs (least privilege)." def check(self, ctx: ProjectContext) -> Iterator[Finding]: + scopes = terraform_scopes(ctx) for file, block in terraform_blocks(ctx): if block.kind == "resource" and block.type in IAM_POLICY_TYPES: text = block.value("policy") or "" line = block.line_of("policy") - statements = hcl_statements(text) + statements = hcl_statements(text, scopes[file.rpartition("/")[0]]) if statements is not None: for statement in statements: if statement.get("Effect") == "Allow" and _has_star( diff --git a/tests/core/test_framework_support.py b/tests/core/test_framework_support.py index 65b7851..12195c8 100644 --- a/tests/core/test_framework_support.py +++ b/tests/core/test_framework_support.py @@ -234,7 +234,7 @@ def test_python_ast_helpers(self) -> None: class TestHclValues: def test_literals_lists_objects_and_jsonencode(self) -> None: - from devshield.core.parsers import Opaque, parse_hcl_value + from devshield.core.parsers import Call, Opaque, parse_hcl_value value = parse_hcl_value( 'jsonencode({\n a = "x"\n "b" : [1, 2.5, true, null]\n c = { d = var.y }\n' @@ -244,7 +244,7 @@ def test_literals_lists_objects_and_jsonencode(self) -> None: "a": "x", "b": [1, 2.5, True, None], "c": {"d": Opaque("var.y")}, - "e": Opaque("merge(...)"), + "e": Call("merge", (Opaque("local.a"), Opaque("local.b"))), "f": Opaque("cond ? 1 : 2"), } @@ -255,3 +255,11 @@ def test_heredoc_json_and_garbage(self) -> None: assert parse_hcl_value("< None: + blocks = parse_hcl( + 'locals {\n a = "x"\n b = [1]\n}\nvariable "v" {\n type = string\n default = "d"\n}\n' + ) + assert [(b.kind, b.labels) for b in blocks] == [("locals", ()), ("variable", ("v",))] + assert blocks[0].value("b") == "[1]" and blocks[1].value("default") == '"d"' diff --git a/tests/rules/test_framework_details.py b/tests/rules/test_framework_details.py index 2637e59..45c14a0 100644 --- a/tests/rules/test_framework_details.py +++ b/tests/rules/test_framework_details.py @@ -109,3 +109,76 @@ def test_config_file_accepts_the_option(self, make_project: MakeProject, value: root = make_project({"devshield.yml": f"scan_dev_files: {value}\n"}) assert load_config(root).scan_dev_files is (value == "true") + + +class TestTerraformReferences: + ADMIN = '{ Effect = "Allow", Action = "*", Resource = "*" }' + + def attach(self, expression: str, extra: str) -> dict[str, str]: + return { + "main.tf": (f'{extra}\nresource "aws_iam_policy" "p" {{\n policy = {expression}\n}}\n') + } + + def severities(self, make_project: MakeProject, files: dict[str, str]) -> list[Severity]: + return [f.severity for f in scan(make_project(files), "SH402").findings] + + def test_locals_are_followed(self, make_project: MakeProject) -> None: + files = self.attach( + "jsonencode(local.doc)", f"locals {{\n doc = {{ Statement = [{self.ADMIN}] }}\n}}" + ) + assert self.severities(make_project, files) == [Severity.CRITICAL] + + def test_a_local_that_holds_the_json_string_is_followed( + self, make_project: MakeProject + ) -> None: + extra = f"locals {{\n doc = jsonencode({{ Statement = [{self.ADMIN}] }})\n}}" + assert self.severities(make_project, self.attach("local.doc", extra)) == [Severity.CRITICAL] + + def test_variable_defaults_are_followed(self, make_project: MakeProject) -> None: + extra = f'variable "statements" {{\n type = any\n default = [{self.ADMIN}]\n}}' + files = self.attach("jsonencode({ Statement = var.statements })", extra) + assert self.severities(make_project, files) == [Severity.CRITICAL] + + def test_concat_and_merge_are_evaluated(self, make_project: MakeProject) -> None: + extra = ( + 'locals {\n safe = [{ Effect = "Allow", Action = "s3:Get*", Resource = "arn:x" }]\n' + f" risky = [{self.ADMIN}]\n" + ' base = { Version = "2012-10-17" }\n}' + ) + files = self.attach( + "jsonencode(merge(local.base, { Statement = concat(local.safe, local.risky) }))", extra + ) + assert self.severities(make_project, files) == [Severity.CRITICAL] + + def test_safe_references_stay_quiet(self, make_project: MakeProject) -> None: + extra = 'locals {\n doc = { Statement = [{ Effect = "Allow", Action = "s3:Get*", Resource = "arn:x" }] }\n}' + assert self.severities(make_project, self.attach("jsonencode(local.doc)", extra)) == [] + + def test_scopes_do_not_leak_between_directories(self, make_project: MakeProject) -> None: + files = { + "a/locals.tf": f"locals {{\n doc = {{ Statement = [{self.ADMIN}] }}\n}}\n", + "b/main.tf": 'resource "aws_iam_policy" "p" {\n policy = jsonencode(local.doc)\n}\n', + } + assert self.severities(make_project, files) == [] + + def test_what_cannot_be_resolved_is_not_guessed(self, make_project: MakeProject) -> None: + for expression in ( + "jsonencode(module.iam.document)", + "data.aws_iam_policy_document.d.json", + "jsonencode(local.missing)", + 'file("policy.json")', + ): + assert self.severities(make_project, self.attach(expression, "")) == [], expression + + def test_self_referencing_locals_terminate(self, make_project: MakeProject) -> None: + extra = "locals {\n a = local.b\n b = local.a\n}" + assert self.severities(make_project, self.attach("local.a", extra)) == [] + + def test_bucket_policies_follow_references_too(self, make_project: MakeProject) -> None: + files = { + "main.tf": ( + 'locals {\n doc = { Statement = [{ Effect = "Allow", Principal = "*", Action = "s3:*" }] }\n}\n' + 'resource "aws_s3_bucket_policy" "p" {\n policy = jsonencode(local.doc)\n}\n' + ) + } + assert len(scan(make_project(files), "SH401").findings) == 1