| 109 | self.replacements.append(spec) |
| 110 | |
| 111 | def request(self, flow: http.HTTPFlow) -> None: |
| 112 | if flow.response or flow.error or not flow.live: |
| 113 | return |
| 114 | |
| 115 | url = flow.request.pretty_url |
| 116 | |
| 117 | all_candidates = [] |
| 118 | for spec in self.replacements: |
| 119 | if spec.matches(flow) and re.search(spec.regex, url): |
| 120 | if spec.local_path.is_file(): |
| 121 | candidates = [spec.local_path] |
| 122 | else: |
| 123 | candidates = file_candidates(url, spec) |
| 124 | all_candidates.extend(candidates) |
| 125 | |
| 126 | local_file = None |
| 127 | for candidate in candidates: |
| 128 | if candidate.is_file(): |
| 129 | local_file = candidate |
| 130 | break |
| 131 | |
| 132 | if local_file: |
| 133 | headers = {"Server": version.MITMPROXY} |
| 134 | mimetype = mimetypes.guess_type(str(local_file))[0] |
| 135 | if mimetype: |
| 136 | headers["Content-Type"] = mimetype |
| 137 | |
| 138 | try: |
| 139 | contents = local_file.read_bytes() |
| 140 | except OSError as e: |
| 141 | logging.warning(f"Could not read file: {e}") |
| 142 | continue |
| 143 | |
| 144 | flow.response = http.Response.make(200, contents, headers) |
| 145 | # only set flow.response once, for the first matching rule |
| 146 | return |
| 147 | if all_candidates: |
| 148 | flow.response = http.Response.make(404) |
| 149 | logging.info( |
| 150 | f"None of the local file candidates exist: {', '.join(str(x) for x in all_candidates)}" |
| 151 | ) |