| 28 | |
| 29 | |
| 30 | class SaveHar: |
| 31 | def __init__(self) -> None: |
| 32 | self.flows: list[flow.Flow] = [] |
| 33 | self.filt: flowfilter.TFilter | None = None |
| 34 | |
| 35 | @command.command("save.har") |
| 36 | def export_har(self, flows: Sequence[flow.Flow], path: types.Path) -> None: |
| 37 | """Export flows to an HAR (HTTP Archive) file.""" |
| 38 | |
| 39 | har = json.dumps(self.make_har(flows), indent=4).encode() |
| 40 | |
| 41 | if path.endswith(".zhar"): |
| 42 | har = zlib.compress(har, 9) |
| 43 | |
| 44 | with open(path, "wb") as f: |
| 45 | f.write(har) |
| 46 | |
| 47 | logging.log(ALERT, f"HAR file saved ({human.pretty_size(len(har))} bytes).") |
| 48 | |
| 49 | def make_har(self, flows: Sequence[flow.Flow]) -> dict: |
| 50 | entries = [] |
| 51 | skipped = 0 |
| 52 | # A list of server seen till now is maintained so we can avoid |
| 53 | # using 'connect' time for entries that use an existing connection. |
| 54 | servers_seen: set[Server] = set() |
| 55 | |
| 56 | for f in flows: |
| 57 | if isinstance(f, http.HTTPFlow): |
| 58 | entries.append(self.flow_entry(f, servers_seen)) |
| 59 | else: |
| 60 | skipped += 1 |
| 61 | |
| 62 | if skipped > 0: |
| 63 | logger.info(f"Skipped {skipped} flows that weren't HTTP flows.") |
| 64 | |
| 65 | return { |
| 66 | "log": { |
| 67 | "version": "1.2", |
| 68 | "creator": { |
| 69 | "name": "mitmproxy", |
| 70 | "version": version.VERSION, |
| 71 | "comment": "", |
| 72 | }, |
| 73 | "pages": [], |
| 74 | "entries": entries, |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | def load(self, loader: Loader): |
| 79 | loader.add_option( |
| 80 | "hardump", |
| 81 | str, |
| 82 | "", |
| 83 | """ |
| 84 | Save a HAR file with all flows on exit. |
| 85 | You may select particular flows by setting save_stream_filter. |
| 86 | For mitmdump, enabling this option will mean that flows are kept in memory. |
| 87 | """, |
no outgoing calls
searching dependent graphs…