Setting the file. One moment.
Resolve Security Model · Security Model · getsentry/skills · Skills Docs
ContentsBack to the top of the page — line 159
This file
Number 24.2
Position 2 of 6
Type Python
Size 17 KB
Lines 404 scripts/ resolve_security_model.py
Python · 404 lines · 17 KB
16 uv run scripts/resolve_security_model.py check --repo PATH [--scope PATH] [--ref REF]
17
18 check reports backticked file paths and code symbols in the policies that no
19 longer exist in the repository, so a policy cannot silently go stale.
20
21 Exit codes: 0 success, 1 check found stale references, 2 error. Warnings go to stderr; results go to stdout.
22 """
23
24 from __future__ import annotations
25
26 import argparse
27 import fnmatch
28 import json
29 import os
30 import re
31 import subprocess
32 import sys
33 import unicodedata
34 from pathlib import Path, PurePosixPath
35
36 DEFAULT_NAME = "SECURITY_MODEL.md"
37 MAX_BYTES = 1024 * 1024 # hard limit, policy is skipped and reported
38 WARN_BYTES = 64 * 1024 # soft limit, large policies crowd out code in context
39 PRUNE_DIRS = { ".git" , "node_modules" , ".venv" , "venv" , "__pycache__" , ".tox" , ".mypy_cache" }
40
41 # Reference check: inline code spans that look like file paths or code symbols.
42 _INLINE_CODE = re.compile( r "` ([ ^` \n ] + ) `" )
43 _SKIP_TOKEN = re.compile( r " [\s =<> \" '#{} \[\] ,;:@|$ \\ ] | :// | ^ -" )
44 _INV_ID = re.compile( r " ^ INV- \d + $ " )
45 _FILE_EXT = re.compile( r " \. ( py | pyi | md | toml | ya ? ml | json | ts | tsx | js | mjs | sh | cfg | ini | txt | sql | lock )$ " )
46 _IDENT = re.compile( r " ^[ A-Za-z_ ][ A-Za-z0-9_- ] * $ " )
47
48 # Characters that change how text renders without being visible: bidi
49 # overrides/isolates, zero-width characters, and other format/control chars.
50 _ALLOWED_CONTROLS = { " \n " , " \r " , " \t " }
51
52
53 class PolicyError ( Exception ):
54 pass
55
56
57 def warn (message: str ) -> None :
58 print ( f "resolve_security_model: warning: { message } " , file = sys.stderr)
59
60
61 def is_hidden (ch: str ) -> bool :
62 if ch in _ALLOWED_CONTROLS :
63 return False
64 return unicodedata.category(ch) in { "Cc" , "Cf" , "Zl" , "Zp" }
65
66
67 def hidden_report (text: str ) -> list[ str ]:
68 findings = []
69 line = 1
70 col = 0
71 for index, ch in enumerate (text):
72 if index == 0 and ch == " \ufeff " : # a leading BOM is harmless
73 continue
74 col += 1
75 if ch == " \n " :
76 line += 1
77 col = 0
78 continue
79 if is_hidden(ch):
80 name = unicodedata.name(ch, "UNNAMED" )
81 findings.append( f "line { line } col { col } : U+ { ord (ch) :04X} { name } " )
82 return findings
83
84
85 def show_hidden (text: str ) -> str :
86 return "" .join( f "<U+ { ord (c) :04X} >" if is_hidden(c) else c for c in text)
87
88
89 def run_git (repo: Path, * args: str , binary: bool = False ) -> bytes | str :
90 try :
91 out = subprocess.run(
92 [ "git" , "-C" , str (repo), * args],
93 check = True ,
94 capture_output = True ,
95 ).stdout
96 except FileNotFoundError as exc:
97 raise PolicyError( "git is not installed" ) from exc
98 except subprocess.CalledProcessError as exc:
99 raise PolicyError(exc.stderr.decode( "utf-8" , "replace" ).strip()) from exc
100 return out if binary else out.decode( "utf-8" )
101
102
103 def git_context (repo: Path) -> tuple[Path, str ] | None :
104 """Return (toplevel, prefix) when repo is inside a git work tree."""
105 try :
106 top = run_git(repo, "rev-parse" , "--show-toplevel" ).strip()
107 prefix = run_git(repo, "rev-parse" , "--show-prefix" ).strip()
108 except PolicyError:
109 return None
110 return Path(top), prefix
111
112
113 def decode_policy (data: bytes , source: str ) -> str :
114 if len (data) > MAX_BYTES :
115 raise PolicyError( f " { source } exceeds 1 MiB; review it manually" )
116 try :
117 text = data.decode( "utf-8" )
118 except UnicodeDecodeError as exc:
119 raise PolicyError( f " { source } is not valid UTF-8" ) from exc
120 if len (data) > WARN_BYTES :
121 warn( f " { source } is { len (data) } bytes; consider condensing it" )
122 for item in hidden_report(text):
123 warn( f " { source } : hidden character at { item } " )
124 return text
125
126
127 class Source :
128 """Reads policy files from the working tree or from a git revision."""
129
130 def __init__ (self, repo: Path, name: str , ref: str | None ):
131 self .repo = repo.resolve()
132 if not self .repo.is_dir():
133 raise PolicyError( f "repo is not a directory: { repo } " )
134 self .name = name
135 self .ref = ref
136 self .git = git_context( self .repo)
137 if ref and not self .git:
138 raise PolicyError( "--ref requires a git repository" )
139 if ref:
140 run_git( self .repo, "rev-parse" , "--verify" , "--quiet" , f " { ref } ^ {{ commit }} " )
141
142 # -- listing -------------------------------------------------------
143 def list (self) -> list[ str ]:
144 if self .git:
145 top, prefix = self .git
146 if self .ref:
147 raw = run_git(top, "ls-tree" , "-r" , "-z" , "--name-only" , self .ref, "--" , prefix or "." , binary = True )
148 else :
149 raw = run_git( self .repo, "ls-files" , "-z" , "-co" , "--exclude-standard" , binary = True )
150 paths = [p.decode( "utf-8" , "surrogateescape" ) for p in raw.split( b " \0 " ) if p]
151 if self .ref and prefix:
152 paths = [p[ len (prefix):] for p in paths if p.startswith(prefix)]
153 found = [p for p in paths if PurePosixPath(p).name == self .name and not self ._disclosure(p)]
154 if not self .ref:
155 found = [p for p in found if self ._safe_worktree_file(p)]
156 return sorted (found)
157 return sorted ( self ._walk())
158
159 def _walk (self) -> list[ str ]:
160 found = []
161 for dirpath, dirnames, filenames in os.walk( self .repo, followlinks = False ):
162 dirnames[:] = [d for d in dirnames if d not in PRUNE_DIRS ]
163 if self .name in filenames:
164 rel = Path(dirpath, self .name).relative_to( self .repo).as_posix()
165 if not self ._disclosure(rel) and self ._safe_worktree_file(rel):
166 found.append(rel)
167 return found
168
169 def _disclosure (self, rel: str ) -> bool :
170 """GitHub treats these SECURITY.md locations as the disclosure policy."""
171 return self .name == "SECURITY.md" and rel in { "SECURITY.md" , ".github/SECURITY.md" , "docs/SECURITY.md" }
172
173 def _safe_worktree_file (self, rel: str ) -> bool :
174 path = self .repo / rel
175 try :
176 real = path.resolve( strict = True )
177 except ( FileNotFoundError , RuntimeError ):
178 return False
179 if not real.is_file():
180 return False
181 if not real.is_relative_to( self .repo):
182 warn( f " { rel } resolves outside the repository; ignored" )
183 return False
184 return True
185
186 # -- reading -------------------------------------------------------
187 def read (self, rel: str ) -> str | None :
188 if self .ref:
189 top, prefix = self .git # type: ignore[misc]
190 spec = f " { self .ref } : { prefix }{ rel } "
191 try :
192 kind = run_git(top, "cat-file" , "-t" , spec).strip()
193 except PolicyError:
194 return None
195 if kind != "blob" :
196 return None
197 return decode_policy(run_git(top, "cat-file" , "blob" , spec, binary = True ), rel)
198 if not self ._safe_worktree_file(rel):
199 return None
200 path = ( self .repo / rel).resolve()
201 with open (path, "rb" ) as fh:
202 data = fh.read( MAX_BYTES + 1 )
203 return decode_policy(data, rel)
204
205 def is_dir (self, rel: str ) -> bool :
206 # A scope missing at this revision (a file the PR adds, or one it
207 # deletes) is treated as a file, so its parent directories still
208 # give the chain. A new directory has no policy of its own there.
209 if self .ref:
210 top, prefix = self .git # type: ignore[misc]
211 target = f " { self .ref } : { prefix }{ rel } " if rel else f " { self .ref } : { prefix } "
212 try :
213 return run_git(top, "cat-file" , "-t" , target).strip() == "tree"
214 except PolicyError:
215 warn( f " { rel } does not exist at { self .ref } ; using its parent directories" )
216 return False
217 path = self .repo / rel
218 if not path.exists():
219 warn( f " { rel } does not exist; using its parent directories" )
220 return False
221 return path.is_dir()
222
223 def relative_scope (self, scope: str ) -> str :
224 raw = Path(os.path.expanduser(scope))
225 absolute = raw if raw.is_absolute() else self .repo / raw
226 resolved = Path(os.path.normpath(absolute))
227 if not self .ref and resolved.exists():
228 resolved = resolved.resolve()
229 if not resolved.is_relative_to( self .repo):
230 raise PolicyError( f "scope is outside the repository: { scope } " )
231 rel = resolved.relative_to( self .repo).as_posix()
232 return "" if rel == "." else rel
233
234 # -- chain ---------------------------------------------------------
235 def resolve (self, scope: str , visible: bool ) -> str :
236 rel = self .relative_scope(scope)
237 target = rel if (rel == "" or self .is_dir(rel)) else str (PurePosixPath(rel).parent)
238 target = "" if target == "." else target
239 parts = PurePosixPath(target).parts if target else ()
240 directories = [ "" ] + [ "/" .join(parts[: i + 1 ]) for i in range ( len (parts))]
241 sections = []
242 for directory in directories:
243 policy_rel = f " { directory } / { self .name } " if directory else self .name
244 if self ._disclosure(policy_rel):
245 continue
246 content = self .read(policy_rel)
247 if content is None or not content.strip():
248 continue
249 if visible:
250 content = show_hidden(content)
251 section = f "## Policy source: { json.dumps(policy_rel) }\n\n{ content } "
252 if not section.endswith( " \n " ):
253 section += " \n "
254 sections.append(section)
255 if not sections:
256 warn( f "no { self .name } applies to { rel or '.' } " )
257 return " \n " .join(sections)
258
259 def chain (self, scope: str ) -> list[ str ]:
260 rel = self .relative_scope(scope)
261 target = rel if (rel == "" or self .is_dir(rel)) else str (PurePosixPath(rel).parent)
262 parts = PurePosixPath(target).parts if target not in { "" , "." } else ()
263 candidates = [ self .name] + [ "/" .join(parts[: i + 1 ]) + f "/ { self .name } " for i in range ( len (parts))]
264 return [c for c in candidates if not self ._disclosure(c) and self .read(c) is not None ]
265
266 # -- reference check -----------------------------------------------
267 def files (self) -> list[ str ]:
268 if self .git:
269 top, prefix = self .git
270 if self .ref:
271 raw = run_git(top, "ls-tree" , "-r" , "-z" , "--name-only" , self .ref, "--" , prefix or "." , binary = True )
272 paths = [p.decode( "utf-8" , "surrogateescape" ) for p in raw.split( b " \0 " ) if p]
273 return [p[ len (prefix):] for p in paths if p.startswith(prefix)]
274 raw = run_git( self .repo, "ls-files" , "-z" , "-co" , "--exclude-standard" , binary = True )
275 return [p.decode( "utf-8" , "surrogateescape" ) for p in raw.split( b " \0 " ) if p]
276 out = []
277 for dirpath, dirnames, filenames in os.walk( self .repo, followlinks = False ):
278 dirnames[:] = [d for d in dirnames if d not in PRUNE_DIRS ]
279 out += [Path(dirpath, f).relative_to( self .repo).as_posix() for f in filenames]
280 return out
281
282 def _text_contains (self, needle: str , word: bool ) -> bool :
283 """Search repository files, excluding policy files, for a reference."""
284 if self .git:
285 args = [ "grep" , "-q" , "-I" , "-F" ] + ([ "-w" ] if word else [])
286 if not self .ref:
287 args.append( "--untracked" )
288 args += [ "-e" , needle]
289 if self .ref:
290 args.append( self .ref)
291 args += [ "--" , "." , f ":(exclude,glob)**/ { self .name } " ]
292 result = subprocess.run([ "git" , "-C" , str ( self .repo), * args], capture_output = True )
293 if result.returncode > 1 :
294 raise PolicyError(result.stderr.decode( "utf-8" , "replace" ).strip())
295 return result.returncode == 0
296 pattern = re.compile(( r " \b %s \b " if word else " %s " ) % re.escape(needle))
297 for rel in self ._file_cache:
298 if PurePosixPath(rel).name == self .name:
299 continue
300 try :
301 with open ( self .repo / rel, encoding = "utf-8" , errors = "ignore" ) as fh:
302 if pattern.search(fh.read( MAX_BYTES )):
303 return True
304 except OSError :
305 continue
306 return False
307
308 def check (self, scope: str | None ) -> list[ str ]:
309 """Report backticked file paths and code symbols that no longer exist."""
310 policies = self .chain(scope) if scope else self .list()
311 self ._file_cache = self .files()
312 file_set = set ( self ._file_cache)
313 problems = []
314 for policy in policies:
315 content = self .read(policy) or ""
316 base = str (PurePosixPath(policy).parent)
317 in_fence = False
318 for lineno, line in enumerate (content.splitlines(), 1 ):
319 if line.lstrip().startswith( "```" ):
320 in_fence = not in_fence
321 continue
322 if in_fence:
323 continue
324 for token in _INLINE_CODE .findall(line):
325 kind, found = self ._check_token(token, base, file_set)
326 if kind and not found:
327 problems.append( f " { policy } : { lineno } : ` { token } ` not found ( { kind } )" )
328 return problems
329
330 def _check_token (self, token: str , base: str , file_set: set[ str ]) -> tuple[ str | None , bool ]:
331 token = token.strip()
332 # Qualified names (`CacheKey::from_scoped_file`) are checked like dotted ones.
333 token = re.sub( r " (?<= \w ) :: (?= \w ) " , "." , token)
334 if _SKIP_TOKEN .search(token) or _INV_ID .match(token) or len (token) < 3 :
335 return None , True
336 token = re.sub( r " \( . * \) $ " , "" , token)
337 if token.startswith( "/" ):
338 return "literal" , self ._text_contains(token, word = False )
339 if "/" in token or _FILE_EXT .search(token):
340 candidates = [token] if base in { "" , "." } else [token, f " { base } / { token } " ]
341 # Authors often write paths relative to a package root
342 # (`api/authentication.py` for `src/sentry/api/authentication.py`),
343 # so a match on the trailing path segments also counts.
344 for cand in candidates:
345 cand = cand.rstrip( "/" )
346 if any (ch in cand for ch in "*?[" ):
347 if any (fnmatch.fnmatchcase(f, cand) or fnmatch.fnmatchcase(f, f "*/ { cand } " ) for f in file_set):
348 return "path" , True
349 elif any (
350 f == cand or f.startswith(cand + "/" ) or f.endswith( "/" + cand) or f "/ { cand } /" in f
351 for f in file_set
352 ):
353 return "path" , True
354 if not self .ref and self ._exists_in_worktree(cand):
355 return "path" , True # reached through a tracked symlink
356 return "path" , False
357 segments = [s for s in token.split( "." ) if len (s) >= 3 ]
358 if not segments or not all ( _IDENT .match(s) for s in segments):
359 return None , True
360 return "symbol" , all ( self ._text_contains(s, word = True ) for s in segments)
361
362 def _exists_in_worktree (self, rel: str ) -> bool :
363 try :
364 real = ( self .repo / rel).resolve( strict = True )
365 except ( OSError , RuntimeError ):
366 return False
367 return real.is_relative_to( self .repo)
368
369
370 def main (argv: list[ str ] | None = None ) -> int :
371 parser = argparse.ArgumentParser( description = __doc__ .split( " \n\n " )[ 0 ], allow_abbrev = False )
372 sub = parser.add_subparsers( dest = "command" , required = True )
373 for cmd in ( "list" , "resolve" , "check" ):
374 p = sub.add_parser(cmd, allow_abbrev = False )
375 p.add_argument( "--repo" , required = True , help = "repository or component root" )
376 p.add_argument( "--ref" , help = "read policies from this git revision (use the PR base for diff reviews)" )
377 p.add_argument( "--name" , default = DEFAULT_NAME , help = f "policy filename (default { DEFAULT_NAME } )" )
378 if cmd == "resolve" :
379 p.add_argument( "--scope" , required = True , help = "file or directory being reviewed" )
380 p.add_argument( "--show-hidden" , action = "store_true" , help = "render hidden characters as <U+XXXX>" )
381 if cmd == "check" :
382 p.add_argument( "--scope" , help = "check only the chain for this path (default: every policy)" )
383 args = parser.parse_args(argv)
384 if "/" in args.name or args.name in { "" , "." , ".." }:
385 parser.error( "--name must be a bare filename" )
386 try :
387 source = Source(Path(os.path.expanduser(args.repo)), args.name, args.ref)
388 if args.command == "list" :
389 print (json.dumps(source.list(), ensure_ascii = True ))
390 elif args.command == "check" :
391 problems = source.check(args.scope)
392 for problem in problems:
393 print (problem)
394 return 1 if problems else 0
395 else :
396 sys.stdout.write(source.resolve(args.scope, args.show_hidden))
397 except PolicyError as exc:
398 print ( f "resolve_security_model: error: { exc } " , file = sys.stderr)
399 return 2
400 return 0
401
402
403 if __name__ == "__main__" :
404 sys.exit(main())