| 49 | |
| 50 | |
| 51 | def save_flows_content(path: pathlib.Path, flows: Iterable[flow.Flow]) -> None: |
| 52 | for f in flows: |
| 53 | assert isinstance(f, http.HTTPFlow) |
| 54 | for m in ("request", "response"): |
| 55 | message = getattr(f, m) |
| 56 | message_path = path / "flows" / f.id / m |
| 57 | os.makedirs(str(message_path / "content"), exist_ok=True) |
| 58 | |
| 59 | with open(str(message_path / "content.data"), "wb") as content_file: |
| 60 | # don't use raw_content here as this is served with a default content type |
| 61 | if message: |
| 62 | content_file.write(message.content) |
| 63 | else: |
| 64 | content_file.write(b"No content.") |
| 65 | |
| 66 | # content_view |
| 67 | t = time.time() |
| 68 | if message: |
| 69 | pretty = contentviews.prettify_message( |
| 70 | message=message, |
| 71 | flow=f, |
| 72 | ) |
| 73 | else: |
| 74 | pretty = contentviews.ContentviewResult( |
| 75 | text="No content.", |
| 76 | syntax_highlight="none", |
| 77 | view_name="/", |
| 78 | description="", |
| 79 | ) |
| 80 | if time.time() - t > 0.1: |
| 81 | logging.info( |
| 82 | f"Slow content view: {pretty.view_name} took {round(time.time() - t, 1)}s", |
| 83 | ) |
| 84 | with (message_path / "content" / "Auto.json").open( |
| 85 | "w" |
| 86 | ) as content_view_file: |
| 87 | json.dump( |
| 88 | dict( |
| 89 | text=pretty.text, |
| 90 | syntax_highlight=pretty.syntax_highlight, |
| 91 | view_name=pretty.view_name, |
| 92 | description=pretty.description, |
| 93 | ), |
| 94 | content_view_file, |
| 95 | ) |
| 96 | |
| 97 | |
| 98 | class StaticViewer: |