Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,8 @@ options:
--only-client record only HTTP client requests
--keep-up, -k keep the server up even if the requests have been read
--force-quit, -q stop the server even if the requests have not been read
--export-html EXPORT_HTML
the path to the export file
--console run a python console (default)
--module MODULE, -m MODULE
run library module as a script (the next args are passed to pytest as is)
Expand Down Expand Up @@ -199,6 +201,10 @@ The requests:
* are still available in the web page even if the python process stopped (except if you force quit before the requests have been loaded by the web page).
* are automatically cleaned if a new execution is detected.

## export

You can export the HTTP traces in a single HTML file. In that case, the web interface is not available during the execution of the python command.

## limitations

Theoretically, if your HTTP client or server uses a standard Python socket, the HTTP requests will be recorded.
Expand Down
21 changes: 17 additions & 4 deletions httpdbg/__init__.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,20 @@
from httpdbg.hooks.all import httprecord
from httpdbg.records import HTTPRecords
__version__ = "2.1.0"

__all__ = ["export_html", "httprecord", "HTTPRecords"]

__version__ = "2.0.2"

__all__ = ["httprecord", "HTTPRecords"]
# lazy loading to avoid circular import issue
def __getattr__(name):
if name == "export_html":
from httpdbg.export import export_html

return export_html
if name == "httprecord":
from httpdbg.hooks.all import httprecord

return httprecord
if name == "HTTPRecords":
from httpdbg.records import HTTPRecords

return HTTPRecords
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
64 changes: 44 additions & 20 deletions httpdbg/__main__.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,19 @@
import argparse
from pathlib import Path
import sys
import time
from typing import Union

from httpdbg import __version__
from httpdbg.args import read_args
from httpdbg.export import export_html
from httpdbg.hooks.all import httprecord
from httpdbg.log import set_env_for_logging
from httpdbg.server import httpdbg_srv
from httpdbg.mode_console import run_console
from httpdbg.mode_module import run_module
from httpdbg.mode_script import run_script
from httpdbg.records import HTTPRecords


def print_msg(msg):
Expand All @@ -17,7 +22,9 @@ def print_msg(msg):
print(msg)


def pyhttpdbg(params, subparams, test_mode=False):
def pyhttpdbg(
params: argparse.Namespace, subparams: list[str], test_mode: bool = False
):

set_env_for_logging(params.log_level, params.log)

Expand All @@ -28,33 +35,50 @@ def pyhttpdbg(params, subparams, test_mode=False):

sys.path.insert(0, "") # to mimic the default python command behavior

with httpdbg_srv(params.host, params.port) as records:
records.server = not params.only_client
with httprecord(records, params.initiator, server=records.server):
def run_recorder(
records: HTTPRecords,
initiators: Union[list[str], None],
record_server: bool,
test_mode: bool,
):
records.server = record_server
with httprecord(records, initiators, server=records.server):
if params.module:
run_module(subparams)
elif params.script:
run_script(subparams)
else:
run_console(records, test_mode)

if not (params.force_quit or test_mode):
if not params.no_banner:
print_msg(f" httpdbg - HTTP(S) requests available at {url}")
if params.export_html:
records = HTTPRecords()

if params.keep_up:
input("Press enter to quit")
else:
# we keep the server up until all the requests have been loaded in the web interface
print(
"Waiting until all the requests have been loaded in the web interface."
)
print("Press Ctrl+C to quit.")
try:
while records.unread:
time.sleep(0.5)
except KeyboardInterrupt: # pragma: no cover
pass
run_recorder(records, params.initiator, not params.only_client, test_mode)

export_html(records, Path(params.export_html))

else:
with httpdbg_srv(params.host, params.port) as records:

run_recorder(records, params.initiator, not params.only_client, test_mode)

if not (params.force_quit or test_mode):
if not params.no_banner:
print_msg(f" httpdbg - HTTP(S) requests available at {url}")

if params.keep_up:
input("Press enter to quit")
else:
# we keep the server up until all the requests have been loaded in the web interface
print(
"Waiting until all the requests have been loaded in the web interface."
)
print("Press Ctrl+C to quit.")
try:
while records.unread:
time.sleep(0.5)
except KeyboardInterrupt: # pragma: no cover
pass


def pyhttpdbg_entry_point(test_mode=False):
Expand Down
12 changes: 9 additions & 3 deletions httpdbg/args.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,22 +40,28 @@ def read_args(args: list[str]) -> tuple[argparse.Namespace, list[str]]:
help="record only HTTP client requests",
)

server_state = parser.add_mutually_exclusive_group()
server_or_export = parser.add_mutually_exclusive_group()

server_state.add_argument(
server_or_export.add_argument(
"--keep-up",
"-k",
action="store_true",
help="keep the server up even if the requests have been read",
)

server_state.add_argument(
server_or_export.add_argument(
"--force-quit",
"-q",
action="store_true",
help="stop the server even if the requests have not been read",
)

server_or_export.add_argument(
"--export-html",
type=str,
help="the path to the export file",
)

parser.add_argument(
"--log-level",
type=LogLevel,
Expand Down
109 changes: 109 additions & 0 deletions httpdbg/export.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
import base64
import json
from pathlib import Path
import re
import xml.etree.ElementTree as ET

import httpdbg
from httpdbg import HTTPRecords
from httpdbg.webapp.api import RequestListPayload, RequestPayload


def generate_html(records: HTTPRecords, for_export: bool = True) -> str:

current_dir = Path(__file__).resolve().parent

with open(Path(current_dir) / "webapp/static/index.htm") as findexhtml:
html = findexhtml.read()

html = html.replace("$**HTTPDBG_VERSION**$", httpdbg.__version__)

# favicon
with open(current_dir / "webapp/static/favicon.ico", "rb") as ffavicon:
b64_icon = base64.b64encode(ffavicon.read()).decode("utf-8")
data_uri = f"data:image/x-icon;base64,{b64_icon}"
html = html.replace(
'<link rel="shortcut icon" href="static/favicon.ico">',
f'<link rel="icon" type="image/x-icon" href="{data_uri}">',
)

# icons
def svg_file_to_symbol(svg_path: Path, symbol_id: str) -> str:
ET.register_namespace("", "http://www.w3.org/2000/svg")
root = ET.parse(svg_path).getroot()
symbol = ET.Element(
"symbol", {"id": symbol_id, "viewBox": root.attrib["viewBox"]}
)
for child in list(root):
symbol.append(child)
return ET.tostring(symbol, encoding="unicode")

icons_inline = ""
for icon_path in (current_dir / "webapp/static/icons").glob("*.svg"):
icons_inline += "\n " + svg_file_to_symbol(
icon_path, icon_path.name[:-4]
)
icons_inline = f'<svg width="0" height="0" style="position:absolute;visibility:hidden" aria-hidden="true" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">{icons_inline}\n </svg>'
html = html.replace("$**PRELOAD_ICONS**$", icons_inline)

# css
stylesheet_pattern = (
r'<link\s+[^>]*rel=["\']stylesheet["\'][^>]*href=["\']([^"\']+)["\']'
)
for stylesheet in re.findall(stylesheet_pattern, html):
with open(Path(current_dir) / "webapp" / stylesheet) as fcss:
html = html.replace(
f'<link rel="stylesheet" href="{stylesheet}">',
f"<style>{fcss.read()}</style>",
)

# js
javascript_pattern = r'<script\s+[^>]*src=["\']([^"\']+)["\']'
for js in re.findall(javascript_pattern, html):
with open(Path(current_dir) / "webapp" / js) as fjs:
html = html.replace(
f'<script src="{js}"></script>', f"<script>{fjs.read()}</script>"
)

# static export of the requests data
if for_export:
static_all_requests = json.dumps(
records, cls=RequestListPayload, ensure_ascii=False
)
map_requests: dict[str, object] = dict()
for record in records:
map_requests[record.id] = json.loads(json.dumps(record, cls=RequestPayload))
static_requests: str = json.dumps(map_requests, ensure_ascii=False)

def safe_for_script_tag(s: str) -> str:
return s.replace("</script>", "<\\/script>")

html_export = f"""
<script id="all-requests" type="application/json">
{safe_for_script_tag(static_all_requests)}
</script>

<script id="requests-map" type="application/json">
{safe_for_script_tag(static_requests)}
</script>

<script>
global.static_all_requests = JSON.parse(
document.getElementById('all-requests').textContent
);
global.static_requests = JSON.parse(
document.getElementById('requests-map').textContent
);
</script>
"""

html = html.replace("$**EXPORT**$", html_export)
else:
html = html.replace("$**EXPORT**$", "")

return html


def export_html(records: HTTPRecords, filename: Path):
with open(filename, "w") as f:
f.write(generate_html(records))
70 changes: 19 additions & 51 deletions httpdbg/webapp/app.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,13 @@
# -*- coding: utf-8 -*-
from collections.abc import Callable
from contextlib import contextmanager
import glob
from http.server import BaseHTTPRequestHandler
import json
import mimetypes
import os
from pathlib import Path
import re
from urllib.parse import ParseResult
from urllib.parse import parse_qs
from urllib.parse import urlparse


from httpdbg import __version__
from httpdbg.log import logger
from httpdbg.webapp.api import RequestListPayload, RequestPayload
from httpdbg.records import HTTPRecords
Expand Down Expand Up @@ -56,61 +51,34 @@ def do_GET(self):
break

def serve_static(self, url: ParseResult):
base_path = os.path.dirname(os.path.realpath(__file__))

if not (
(url.path.lower() in ["/", "index.htm", "index.html"])
or url.path.startswith("/static/")
):
return False

if url.path.lower() in ["/", "index.htm", "index.html"]:
self.path = "/static/index.htm"

# get real path
self.path = self.path.split("-+-")[0]

fullpath = os.path.realpath(os.path.join(base_path, self.path[1:]))
from httpdbg.export import generate_html

if not fullpath.startswith(base_path):
# if the file is not in the static directory, we don't serve it
return self.serve_not_found(url)

if not os.path.exists(fullpath):
return self.serve_not_found(url)
else:
self.send_response(200)
self.send_header(
"content-type", mimetypes.types_map[os.path.splitext(fullpath)[1]]
)
if self.path == "/static/index.htm":
self.send_header_no_cache()
else:
self.send_header_with_cache(604800)
self.send_header("content-type", "text/html")
self.send_header_no_cache()
self.end_headers()
self.wfile.write(
generate_html(self.records, for_export=False).encode("utf-8")
)
return True

with open(fullpath, "rb") as f:
filecontent = f.read()

if b"$**PRELOAD_ICONS**$" in filecontent:
icons = ""
for icon in glob.glob(
os.path.realpath(base_path) + "/static/icons/*.svg"
):
icon_path = icon.replace(os.path.realpath(base_path) + "/", "")
icons += f" <link rel='preload' href='{icon_path}-+-$**HTTPDBG_VERSION**$' as='image' type='image/svg+xml' />\n"
if url.path.lower() == "favicon.ico":
self.send_response(200)
self.send_header("content-type", "image/x-icon")
self.send_header_no_cache()
self.end_headers()

filecontent = filecontent.replace(
b"$**PRELOAD_ICONS**$", icons.encode("utf-8")
)
current_dir = Path(__file__).resolve().parent

filecontent = filecontent.replace(
b"$**HTTPDBG_VERSION**$", __version__.encode("utf-8")
)
with open(Path(current_dir) / "static/favicon.ico") as f:
filecontent = f.read()
self.wfile.write(filecontent.encode("utf-8"))

self.wfile.write(filecontent)
return True

return True
return False

def serve_requests(self, url: ParseResult):
if not (url.path.lower() == "/requests"):
Expand Down
Loading