Do not follow a .py symlink out of the scan root

CHANGELOG.md

@@ -8,6 +8,16 @@ All notable changes to CodeSnake are documented here. The format follows
 
 ### Security
 
+- **A directory walk no longer follows a `.py` symlink out of the scan root.**
+  `os.walk` does not follow directory symlinks, but a *file* symlink was yielded and
+  read through, so `src/x.py -> ~/.ssh/id_rsa` was opened when scanning `src/`. Worse
+  than the read itself, a syntax error prints the offending source line into the
+  report, so a line of the target file reached the output — and on a public repository,
+  the CI log. Symlinks resolving inside the root are still followed; an explicit file
+  argument is still read as given.
+
+### Security
+
 - **The release workflow refuses a tag that is not reachable from `main`.** Branch
   protection governs `main`, not tags, so anyone able to push a tag and publish a
   GitHub Release could previously ship a commit that never passed review — the

SECURITY.md

@@ -34,6 +34,8 @@ Two places reach outside the process, both only when you ask:
 
 Directory walks skip `.gitignore` matches, but **only for files git does not track** — matching git's own rule that an ignore pattern has no effect on a committed file. A file added with `git add -f` is therefore still analyzed, and so is a tracked file inside an ignored directory. `--no-ignore` disables ignore handling entirely, for scanning a working tree that is not a git repository or whose ignored output you want covered. Venvs, caches, and `.git` are skipped in every mode.
 
+A directory walk does not follow a `.py` symlink out of the tree being scanned. `src/x.py -> ~/.ssh/id_rsa` is skipped rather than read, because a parse failure prints the offending source line into the report. Symlinks that resolve inside the scan root are followed normally. An explicit file argument is still read as given — naming a file is a deliberate act.
+
 ### In scope
 
 - Anything that makes CodeSnake execute code from a file it is analyzing

src/codesnake/checker.py

@@ -432,6 +432,19 @@
     return code.upper() in codes
 
 
+def _within(path: Path, root: Path) -> bool:
+    """True if ``path`` resolves to somewhere inside ``root``.
+
+    Reading through a symlink that leaves the tree the caller asked to scan
+    discloses files outside it, so the walk declines to follow one.
+    """
+    try:
+        resolved = path.resolve()
+    except OSError:
+        return False
+    return resolved == root or resolved.is_relative_to(root)
+
+
 def _load_ignore_state(
     root_resolved: Path,
 ) -> Tuple['_IgnoreStack', Set[Path], Set[Path]]:
@@ -470,54 +483,60 @@
     return ignore, tracked, tracked_dirs
 
 
 def iter_python_files(root: Path, *, respect_gitignore: bool = True) -> Iterable[Path]:
     """Yield .py files under root, skipping venvs, caches, and (by default) .gitignore matches.
 
     ``respect_gitignore=False`` still skips ``SKIP_DIR_NAMES`` (venvs, caches,
     ``.git``, ...) so a CI gate does not have to crawl ``site-packages``. It
     does not skip a committed file that ``.gitignore`` would hide — pass
     ``--no-ignore`` for that, or name the file explicitly.
     """
     try:
         root_resolved = root.resolve()
     except OSError:
         root_resolved = root
     ignore, tracked, tracked_dirs = (
         _load_ignore_state(root_resolved) if respect_gitignore
         else (_IgnoreStack(), set(), set())
     )
 
     for dirpath, dirnames, filenames in os.walk(root_resolved):
         current = Path(dirpath)
         views: _DirectoryView = []
         if respect_gitignore:
             nested_gi = current / '.gitignore'
             if nested_gi.is_file():
                 ignore.add_gitignore(nested_gi)
             views = ignore.view(current)
 
         dirnames[:] = [
             name for name in dirnames
             if name not in SKIP_DIR_NAMES
             and not name.endswith('.egg-info')
             and not (
                 respect_gitignore
                 and (current / name) not in tracked_dirs
                 and _IgnoreStack.ignored(views, name, is_dir=True)
             )
         ]
 
         for name in filenames:
             if not name.endswith('.py'):
                 continue
             candidate = current / name
+            # os.walk does not follow directory symlinks, but a *file* symlink is
+            # yielded and would be read through. `src/x.py -> ~/.ssh/id_rsa` would
+            # otherwise be opened, and a syntax error prints the offending line
+            # into the report. Only follow links that stay inside the scan root.
+            if candidate.is_symlink() and not _within(candidate, root_resolved):
+                continue
             if (
                 respect_gitignore
                 and candidate not in tracked
                 and _IgnoreStack.ignored(views, name, is_dir=False)
             ):
                 continue
             yield candidate
 
 
 @dataclass
@@ -706,87 +725,87 @@
     return CheckerConfig.from_file(str(found))
 
 
 class SemanticChecker(ast.NodeVisitor):
     """Main semantic checker that analyzes Python AST for issues."""
 
     def __init__(
         self,
         source_code: str,
         filename: str = '<string>',
         config: Optional[CheckerConfig] = None,
         known_exports: Optional[Dict[str, Set[str]]] = None,
     ):
         self.source_code = source_code
         self.filename = filename
         self.config = config or CheckerConfig()
         self.issues: List[Issue] = []
         self.source_lines = source_code.split('\n')
 
         self.aliases: Dict[str, str] = {}
         # Cyclomatic complexity per function name (nested scopes not charged).
         self.function_complexity: Dict[str, int] = {}
         self.scopes: List[_Scope] = []
         self._in_type_checking = False
         self._with_expr_ids: Set[int] = set()
         # id(Name node) -> binding kind for stores that are not plain assignments
         # (loop targets, tuple unpacking, bare annotations).
         self._store_kinds: Dict[int, str] = {}
         # Function bodies are analyzed after the enclosing scope is fully bound,
         # so closures may reference names assigned later in that scope.
         self._deferred: List[Tuple[ast.AST, List[_Scope], bool]] = []
         self.known_exports = known_exports or {}
 
     def add_issue(
         self,
         severity: str,
         category: str,
         message: str,
         node: ast.AST,
         code: str,
     ):
         """Add an issue at ``node``'s location if its category is enabled."""
         self._record_issue(
             severity,
             category,
             message,
             getattr(node, 'lineno', 0) or 0,
             getattr(node, 'col_offset', 0) or 0,
             code,
             end_line=getattr(node, 'end_lineno', None) or getattr(node, 'lineno', 0) or 0,
             end_col=getattr(node, 'end_col_offset', None) or getattr(node, 'col_offset', 0) or 0,
         )
 
     def _record_issue(
         self,
         severity: str,
         category: str,
         message: str,
         line: int,
         col: int,
         code: str,
         source: str = 'codesnake',
         end_line: int = 0,
         end_col: int = 0,
         suggestion: str = '',
     ) -> None:
         if not self.config.allows_category(category):
             return
         end_line = end_line or line
         col = self._char_col(line, col)
         end_col = self._char_col(end_line, end_col) if end_col else col
         self.issues.append(Issue(
             severity=severity,
             category=category,
             message=message,
             line=line,
             col=col,
             code=code,
             filename=self.filename,
             source=source,
             end_line=end_line,
             end_col=end_col,
             suggestion=suggestion or ISSUE_SUGGESTIONS.get(code, ''),
         ))
 
     def _char_col(self, line: int, col: int) -> int:
         """Convert an AST UTF-8 byte offset into a 0-based character offset."""
@@ -815,30 +834,30 @@
             self._report_unused_imports(scope)
         return scope
 
     def _bind(self, name: str, node: ast.AST, kind: str) -> None:
         scope = self._current_scope()
         if scope is None:
             return
         if name in scope.global_names:
             if len(self.scopes) > 1:
                 self.scopes[0].bindings.setdefault(name, _Binding(
                     name,
                     getattr(node, 'lineno', 0) or 0,
                     getattr(node, 'col_offset', 0) or 0,
                     kind,
                 ))
             return
         if name in scope.nonlocal_names:
             return
         if name not in scope.bindings:
             if kind in _SHADOW_CHECKED_KINDS and scope.kind == 'function':
                 self._maybe_shadow(name, node)
             scope.bindings[name] = _Binding(
                 name,
                 getattr(node, 'lineno', 0) or 0,
                 getattr(node, 'col_offset', 0) or 0,
                 kind,
             )
 
     def _maybe_shadow(self, name: str, node: ast.AST) -> None:
         if name.startswith('_') or name in SKIP_UNUSED_NAMES or name in _BUILTIN_NAMES:
@@ -909,45 +928,45 @@
         elif isinstance(target, ast.Starred):
             self._mark_tainted_target(target.value)
 
     def _is_tainted_expr(self, node: Optional[ast.AST]) -> bool:
         if node is None:
             return False
         if isinstance(node, ast.Constant):
             return False
         if isinstance(node, ast.Name):
             return self._name_is_tainted(node.id)
         if isinstance(node, ast.JoinedStr):
             return any(self._is_tainted_expr(value) for value in node.values)
         if isinstance(node, ast.FormattedValue):
             return self._is_tainted_expr(node.value)
         if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Mod)):
             return self._is_tainted_expr(node.left) or self._is_tainted_expr(node.right)
         if isinstance(node, ast.Call):
             resolved = self._resolve_name(node.func)
             if resolved in TAINT_CALL_NAMES:
                 return True
             if resolved in SANITIZER_NAMES:
                 return False
             if isinstance(node.func, ast.Attribute) and node.func.attr in ('get', 'format'):
                 if self._is_tainted_expr(node.func.value):
                     return True
                 if self._call_args_tainted(node):
                     return True
             return self._call_args_tainted(node)
         if isinstance(node, ast.Attribute):
             resolved = self._resolve_name(node)
             if resolved in TAINT_ATTR_NAMES:
                 return True
             if node.attr in REQUEST_TAINT_ATTRS and self._is_request_like(node.value):
                 return True
             return self._is_tainted_expr(node.value)
         if isinstance(node, ast.Subscript):
             return self._is_tainted_expr(node.value)
         if isinstance(node, ast.Starred):
             return self._is_tainted_expr(node.value)
         if isinstance(node, (ast.List, ast.Tuple)):
             return any(self._is_tainted_expr(elt) for elt in node.elts)
         return False
 
     @staticmethod
     def _first_arg(node: ast.Call, keyword: Optional[str] = None) -> Optional[ast.AST]:
@@ -1099,21 +1118,21 @@
         if isinstance(node.target, ast.Name) and node.target.id == '__all__':
             self._mark_exported(node.value)
 
     def _record_dunder_all_call(self, node: ast.Call) -> None:
         """``__all__.extend([...])`` / ``__all__.append('name')``."""
         if not self._at_module_scope():
             return
         func = node.func
         if not isinstance(func, ast.Attribute):
             return
         if not isinstance(func.value, ast.Name) or func.value.id != '__all__':
             return
         if func.attr == 'extend' and node.args:
             self._mark_exported(node.args[0])
         elif func.attr == 'append':
             for arg in node.args:
                 if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                     self.scopes[0].used.add(arg.value)
 
     def _record_constant_assign(self, targets: Sequence[ast.AST], value: ast.AST) -> None:
         if not isinstance(value, ast.Constant):
@@ -1122,26 +1141,26 @@
             if isinstance(target, ast.Name):
                 self._set_const(target.id, value.value)
 
     def _collect_imports(self, tree: ast.AST) -> None:
         """Build a name -> qualified-name map from import statements."""
         for node in ast.walk(tree):
             if isinstance(node, ast.Import):
                 for alias in node.names:
                     if alias.asname:
                         self.aliases[alias.asname] = alias.name
                     else:
                         root = alias.name.split('.')[0]
                         self.aliases[root] = root
             elif isinstance(node, ast.ImportFrom):
                 module = node.module or ''
                 for alias in node.names:
                     if alias.name == '*':
                         continue
                     local = alias.asname or alias.name
                     if module:
                         self.aliases[local] = f'{module}.{alias.name}'
                     else:
                         self.aliases[local] = alias.name
 
     def _resolve_name(self, node: Optional[ast.AST]) -> Optional[str]:
         """Return a dotted name for a Call target, using the import map."""
@@ -1166,27 +1185,27 @@
                 return True
         return False
 
     def _is_stub_body(self, node: ast.AST) -> bool:
         body = list(getattr(node, 'body', []))
         if (
             body
             and isinstance(body[0], ast.Expr)
             and isinstance(body[0].value, ast.Constant)
             and isinstance(body[0].value.value, str)
         ):
             body = body[1:]
         if not body:
             return True
         if len(body) == 1 and isinstance(body[0], ast.Pass):
             return True
         if (
             len(body) == 1
             and isinstance(body[0], ast.Expr)
             and isinstance(body[0].value, ast.Constant)
             and body[0].value.value is ...
         ):
             return True
         return False
 
     def _async_body_has_await(self, node: ast.AST) -> bool:
         for child in ast.iter_child_nodes(node):

test/test_codesnake.py

@@ -881,6 +881,64 @@ def _git(self, root, *args):
         import subprocess
         subprocess.run(['git', *args], cwd=root, check=True, capture_output=True, timeout=30)
 
+    def test_symlink_escaping_the_root_is_not_followed(self):
+        """`src/x.py -> /outside/secret` must not be read.
+
+        os.walk does not follow directory symlinks, but a file symlink is
+        yielded and would be read through, and a syntax error prints the
+        offending line into the report.
+        """
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            (root / 'project' / 'src').mkdir(parents=True)
+            outside = root / 'outside'
+            outside.mkdir()
+            secret = outside / 'secret.txt'
+            secret.write_text('DB_PASSWORD = hunter2!$nope\n', encoding='utf-8')
+            (root / 'project' / 'src' / 'real.py').write_text('x = 1\n', encoding='utf-8')
+            try:
+                (root / 'project' / 'src' / 'leaked.py').symlink_to(secret)
+            except (OSError, NotImplementedError):
+                self.skipTest('symlinks unavailable')
+
+            targets, _ = expand_python_targets([str(root / 'project' / 'src')])
+            self.assertEqual({Path(t).name for t in targets}, {'real.py'})
+
+    def test_symlink_staying_inside_the_root_is_followed(self):
+        """Containment, not a blanket ban on symlinks."""
+        from codesnake.checker import iter_python_files
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            (root / 'src' / 'pkg').mkdir(parents=True)
+            target = root / 'src' / 'pkg' / 'shared.py'
+            target.write_text('x = 1\n', encoding='utf-8')
+            try:
+                (root / 'src' / 'alias.py').symlink_to(target)
+            except (OSError, NotImplementedError):
+                self.skipTest('symlinks unavailable')
+
+            found = {p.name for p in iter_python_files(root / 'src')}
+            self.assertEqual(found, {'shared.py', 'alias.py'})
+
+    def test_escaping_symlink_content_never_reaches_the_report(self):
+        """The disclosure that made this worth fixing: the source line is printed."""
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            (root / 'project' / 'src').mkdir(parents=True)
+            secret = root / 'secret.env'
+            secret.write_text('DEBUG = False\nDB_PASSWORD = hunter2!$nope\n', encoding='utf-8')
+            (root / 'project' / 'src' / 'ok.py').write_text('x = 1\n', encoding='utf-8')
+            try:
+                (root / 'project' / 'src' / 'config.py').symlink_to(secret)
+            except (OSError, NotImplementedError):
+                self.skipTest('symlinks unavailable')
+
+            out = StringIO()
+            run_check([str(root / 'project' / 'src')], stream=out,
+                      color=False, show_banner=False)
+            self.assertNotIn('DB_PASSWORD', out.getvalue())
+            self.assertNotIn('hunter2', out.getvalue())
+
     def test_committed_file_is_not_hidden_by_gitignore(self):
         """git ignores .gitignore for tracked files; so must the walk.