|
| 1 | +"""Offline scanning for cloned repositories and local source trees.""" |
| 2 | + |
| 3 | +import shlex |
| 4 | +import subprocess |
| 5 | +import time |
| 6 | +from contextlib import nullcontext |
| 7 | +from dataclasses import dataclass |
| 8 | +from pathlib import Path |
| 9 | + |
| 10 | +from github_dorks.dictionaries import iter_dorks, resolve_dictionaries |
| 11 | +from github_dorks.output import create_writer |
| 12 | + |
| 13 | + |
| 14 | +LANGUAGE_EXTENSIONS = { |
| 15 | + 'bash': {'.bash', '.sh'}, |
| 16 | + 'json': {'.json'}, |
| 17 | + 'php': {'.php'}, |
| 18 | + 'python': {'.py'}, |
| 19 | + 'ruby': {'.rb'}, |
| 20 | + 'shell': {'.bash', '.fish', '.ksh', '.sh', '.zsh'}, |
| 21 | + 'yaml': {'.yaml', '.yml'}, |
| 22 | +} |
| 23 | +QUALIFIERS = {'extension', 'filename', 'language', 'path'} |
| 24 | + |
| 25 | + |
| 26 | +@dataclass |
| 27 | +class LocalScanStats: |
| 28 | + queries: int = 0 |
| 29 | + matches: int = 0 |
| 30 | + files: int = 0 |
| 31 | + failures: int = 0 |
| 32 | + retries: int = 0 |
| 33 | + elapsed_seconds: float = 0.0 |
| 34 | + |
| 35 | + @property |
| 36 | + def exit_code(self): |
| 37 | + return 2 if self.failures else 0 |
| 38 | + |
| 39 | + |
| 40 | +@dataclass(frozen=True) |
| 41 | +class Predicate: |
| 42 | + kind: str |
| 43 | + value: str |
| 44 | + |
| 45 | + def matches(self, path, relative_path, normalized_text): |
| 46 | + value = self.value.casefold() |
| 47 | + if self.kind == 'filename': |
| 48 | + return value in path.name.casefold() |
| 49 | + if self.kind == 'extension': |
| 50 | + return path.suffix.casefold().lstrip('.') == value.lstrip('.') |
| 51 | + if self.kind == 'path': |
| 52 | + return value in relative_path.casefold() |
| 53 | + if self.kind == 'language': |
| 54 | + return path.suffix.casefold() in LANGUAGE_EXTENSIONS.get(value, set()) |
| 55 | + return value in normalized_text |
| 56 | + |
| 57 | + |
| 58 | +@dataclass(frozen=True) |
| 59 | +class LocalQuery: |
| 60 | + source: str |
| 61 | + clauses: tuple |
| 62 | + exclusions: tuple |
| 63 | + |
| 64 | + def matches(self, path, relative_path, normalized_text): |
| 65 | + if any( |
| 66 | + predicate.matches(path, relative_path, normalized_text) |
| 67 | + for predicate in self.exclusions |
| 68 | + ): |
| 69 | + return False |
| 70 | + return all( |
| 71 | + any( |
| 72 | + predicate.matches(path, relative_path, normalized_text) |
| 73 | + for predicate in clause |
| 74 | + ) |
| 75 | + for clause in self.clauses |
| 76 | + ) |
| 77 | + |
| 78 | + def snippets(self, text, limit=5): |
| 79 | + needles = { |
| 80 | + predicate.value.casefold() |
| 81 | + for clause in self.clauses |
| 82 | + for predicate in clause |
| 83 | + if predicate.kind == 'content' |
| 84 | + } |
| 85 | + if not needles: |
| 86 | + return [] |
| 87 | + matches = [] |
| 88 | + for number, line in enumerate(text.splitlines(), 1): |
| 89 | + if any(needle in line.casefold() for needle in needles): |
| 90 | + matches.append(f'{number}: {line.strip()}'[:300]) |
| 91 | + if len(matches) == limit: |
| 92 | + break |
| 93 | + return matches |
| 94 | + |
| 95 | + |
| 96 | +def parse_query(query): |
| 97 | + """Parse the useful local subset of GitHub code-search syntax.""" |
| 98 | + tokens = shlex.split(query) |
| 99 | + clauses = [] |
| 100 | + exclusions = [] |
| 101 | + join_previous = False |
| 102 | + negate_next = False |
| 103 | + |
| 104 | + for token in tokens: |
| 105 | + operator = token.upper() |
| 106 | + if operator == 'OR': |
| 107 | + join_previous = True |
| 108 | + continue |
| 109 | + if operator == 'NOT': |
| 110 | + negate_next = True |
| 111 | + continue |
| 112 | + |
| 113 | + name, separator, value = token.partition(':') |
| 114 | + predicate = Predicate( |
| 115 | + name.casefold() if separator and name.casefold() in QUALIFIERS |
| 116 | + else 'content', |
| 117 | + value if separator and name.casefold() in QUALIFIERS else token, |
| 118 | + ) |
| 119 | + if negate_next: |
| 120 | + exclusions.append(predicate) |
| 121 | + negate_next = False |
| 122 | + join_previous = False |
| 123 | + elif join_previous and clauses: |
| 124 | + clauses[-1].append(predicate) |
| 125 | + join_previous = False |
| 126 | + else: |
| 127 | + clauses.append([predicate]) |
| 128 | + |
| 129 | + if negate_next or join_previous: |
| 130 | + raise ValueError(f'incomplete operator in dork: {query}') |
| 131 | + return LocalQuery( |
| 132 | + source=query, |
| 133 | + clauses=tuple(tuple(clause) for clause in clauses), |
| 134 | + exclusions=tuple(exclusions), |
| 135 | + ) |
| 136 | + |
| 137 | + |
| 138 | +def _git_files(root): |
| 139 | + try: |
| 140 | + result = subprocess.run( |
| 141 | + [ |
| 142 | + 'git', '-C', str(root), 'ls-files', '-z', '--cached', '--others', |
| 143 | + '--exclude-standard', |
| 144 | + ], |
| 145 | + stdout=subprocess.PIPE, |
| 146 | + stderr=subprocess.DEVNULL, |
| 147 | + check=False, |
| 148 | + ) |
| 149 | + except OSError: |
| 150 | + return None |
| 151 | + if result.returncode: |
| 152 | + return None |
| 153 | + return [ |
| 154 | + root / name |
| 155 | + for name in result.stdout.decode(errors='surrogateescape').split('\0') |
| 156 | + if name |
| 157 | + ] |
| 158 | + |
| 159 | + |
| 160 | +def iter_local_files(target): |
| 161 | + """Yield Git-tracked and non-ignored files, with a filesystem fallback.""" |
| 162 | + target = Path(target).resolve() |
| 163 | + if target.is_file(): |
| 164 | + yield target, target.name |
| 165 | + return |
| 166 | + if not target.is_dir(): |
| 167 | + raise ValueError(f'local scan path is not a file or directory: {target}') |
| 168 | + |
| 169 | + paths = _git_files(target) |
| 170 | + if paths is None: |
| 171 | + paths = ( |
| 172 | + path for path in target.rglob('*') |
| 173 | + if '.git' not in path.parts |
| 174 | + ) |
| 175 | + for path in paths: |
| 176 | + if path.is_file() and not path.is_symlink(): |
| 177 | + yield path, path.relative_to(target).as_posix() |
| 178 | + |
| 179 | + |
| 180 | +def _read_text(path, max_file_size): |
| 181 | + if path.stat().st_size > max_file_size: |
| 182 | + return None |
| 183 | + contents = path.read_bytes() |
| 184 | + if b'\0' in contents: |
| 185 | + return None |
| 186 | + return contents.decode('utf-8', errors='replace') |
| 187 | + |
| 188 | + |
| 189 | +def scan_local(local_path, gh_dorks_file=None, categories=None, |
| 190 | + output_filename=None, output_format=None, force=False, |
| 191 | + quiet=False, verbose=False, max_file_size=1_000_000, |
| 192 | + monotonic=time.monotonic, stdout=None, stderr=None): |
| 193 | + """Scan a local working tree with the bundled or custom dorks.""" |
| 194 | + import sys |
| 195 | + |
| 196 | + if max_file_size <= 0: |
| 197 | + raise ValueError('maximum file size must be greater than zero') |
| 198 | + stdout = sys.stdout if stdout is None else stdout |
| 199 | + stderr = sys.stderr if stderr is None else stderr |
| 200 | + root = Path(local_path).resolve() |
| 201 | + excluded_output = Path(output_filename).resolve() if output_filename else None |
| 202 | + sources = resolve_dictionaries(gh_dorks_file, categories) |
| 203 | + queries = [parse_query(dork) for dork in iter_dorks(sources)] |
| 204 | + output_format = output_format or ('csv' if output_filename else 'text') |
| 205 | + stats = LocalScanStats(queries=len(queries)) |
| 206 | + started_at = monotonic() |
| 207 | + status_stream = stderr if not output_filename and output_format != 'text' else stdout |
| 208 | + if not quiet: |
| 209 | + status_stream.write(f'Scanning local path: {root}\n') |
| 210 | + |
| 211 | + output_context = ( |
| 212 | + open(output_filename, 'w' if force else 'x', newline='', encoding='utf-8') |
| 213 | + if output_filename else nullcontext(stdout) |
| 214 | + ) |
| 215 | + with output_context as output_file: |
| 216 | + writer = create_writer(output_format, output_file) |
| 217 | + writer.start() |
| 218 | + try: |
| 219 | + for path, relative_path in iter_local_files(root): |
| 220 | + if excluded_output and path.resolve() == excluded_output: |
| 221 | + continue |
| 222 | + try: |
| 223 | + text = _read_text(path, max_file_size) |
| 224 | + except OSError as error: |
| 225 | + if verbose and not quiet: |
| 226 | + status_stream.write(f'Skipping {relative_path}: {error}\n') |
| 227 | + continue |
| 228 | + if text is None: |
| 229 | + continue |
| 230 | + stats.files += 1 |
| 231 | + normalized_text = text.casefold() |
| 232 | + for query in queries: |
| 233 | + if query.matches(path, relative_path, normalized_text): |
| 234 | + stats.matches += 1 |
| 235 | + writer.write({ |
| 236 | + 'dork': query.source, |
| 237 | + 'text_matches': query.snippets(text), |
| 238 | + 'path': relative_path, |
| 239 | + 'score': len(query.clauses), |
| 240 | + 'url': path.as_uri(), |
| 241 | + }) |
| 242 | + finally: |
| 243 | + stats.elapsed_seconds = monotonic() - started_at |
| 244 | + writer.finish(stats) |
| 245 | + |
| 246 | + if not quiet: |
| 247 | + if not stats.matches: |
| 248 | + status_stream.write('No local matches found. Hurray!\n') |
| 249 | + status_stream.write( |
| 250 | + f'Summary: {stats.queries} queries, {stats.matches} matches, ' |
| 251 | + f'{stats.files} files, {stats.elapsed_seconds:.1f}s elapsed\n' |
| 252 | + ) |
| 253 | + return stats |
0 commit comments