xepor package

class xepor.InterceptedAPI(default_host=None, host_mapping={}, blacklist_domain=[], request_passthrough=True, response_passthrough=True, respect_proxy_headers=False)[source]

Bases: object

the InterceptedAPI object is the central registry of your view functions. Users should use a function decorator route() to define and register URL and host mapping to the view functions. Just like flask’s flask.Flask.route().

from xepor import InterceptedAPI, RouteType

HOST_HTTPBIN = "httpbin.org"
api = InterceptedAPI(HOST_HTTPBIN)

Defining a constant for your target (victim) domain name is not mandatory (even the default_host parameter itself is optional) but recommanded as a best practise. If you have multiple hosts to inject (see an example at xepor/xepor-examples/polyv_scrapper/polyv.py), you would have to specify the domain name multiple times in each route() in host parameter, (especially for domains other than default_host). So it’s better to have a variable for that.

Add route via function call similar to Flask flask.Flask.add_url_rule() is not yet implemented.

Parameters:

  • default_host (str | None) – The default host to forward requests to.

  • host_mapping (List[Tuple[str | Pattern, str]]) – A list of tuples of the form (regex, host) where regex is a regular expression to match against the request host and host is the host to redirect the request to.

  • blacklist_domain (List[str]) – A list of domains to not forward requests to. The requests and response from hosts in this list will not respect request_passthrough and response_passthrough setting.

  • request_passthrough (bool) – Whether to forward the request to upstream server if no route is found. If request_passthrough = False, all requests not matching any route will be responded with default_response() without connecting to upstream.

  • response_passthrough (bool) – Whether to forward the response to the user if no route is found. If response_passthrough = False, all responses not matching any route will be replaced with the Response object generated by default_response().

  • respect_proxy_headers (bool) –

    Set to True only when you use Xepor as a web server behind a reverse proxy. Typical use case is to set up an mitmproxy in reverse mode to bypass some online license checks. Xepor will respect the following headers and strip them from requests to upstream.

    • X-Forwarded-For

    • X-Forwarded-Host

    • X-Forwarded-Port

    • X-Forwarded-Proto

    • X-Forwarded-Server

    • X-Real-Ip

load(loader)[source]

This function is called by the mitmproxy framework before proxy server started. Currently it’s used to set a must-have mitmproxy option for Xepor to work: connection_strategy=lazy. If you want to override this method, remember to call super().load(loader) in your code.

User can also import and use mitmproxy.ctx object to configure other options for mitmproxy when overriding this function.

from mitmproxy import ctx

ctx.options.connection_strategy = "lazy"
Parameters:

loader (Loader) – a mitmproxy.addonmanager.Loader which can be used to add custom options.

Returns:

None

request(flow)[source]

This function is called by the mitmproxy framework whenever a request is made.

Parameters:

flow (HTTPFlow) – The mitmproxy.http.HTTPFlow object from client request.

Returns:

None

response(flow)[source]

This function is called by the mitmproxy when a response is returned the server.

Parameters:

flow (HTTPFlow) – The mitmproxy.http.HTTPFlow object from server response.

Returns:

None

route(path, host=None, rtype=RouteType.REQUEST, catch_error=True, return_error=False)[source]

This is the main API used by end users. It decorate a view function to register it with given host and URL.

Typical usage (taken from official example: httpbin.py):

@api.route("/get")
def change_your_request(flow: HTTPFlow):
    flow.request.query["payload"] = "evil_param"

@api.route("/basic-auth/{usr}/{pwd}", rtype=RouteType.RESPONSE)
def capture_auth(flow: HTTPFlow, usr=None, pwd=None):
    print(
        f"auth @ {usr} + {pwd}:",
        f"Captured {'successful' if flow.response.status_code < 300 else 'unsuccessful'} login:",
        flow.request.headers.get("Authorization", ""),
    )

See Github: xepor/xepor-examples for more examples.

Parameters:

  • path (str) – The URL path to be routed. The path definition grammar is similar to Python 3 str.format(). Check the documentation of parse library: r1chardj0n3s/parse

  • host (str | None) –

    The host to be routed. This value will be matched against the following fields of incoming flow object by order:

    1. X-Forwarded-For Header. (only when respect_proxy_headers in InterceptedAPI is True)

    2. HTTP Host Header, if exists.

    3. flow.host reported by underlying layer. In HTTP or Socks5h proxy mode, it may hopefully be a hostname, otherwise, it’ll be an IP address.

  • rtype (RouteType) – Set the route be matched on either request or response. Accepting RouteType.

  • catch_error (bool) –

    If set to True, the exception inside the route will be handled by Xepor.

    If set to False, the exception will be raised and handled by mitmproxy.

  • return_error (bool) –

    If set to True, the error message inside the exception (str(exc)) will be returned to client. This behaviour can be overrided through error_response().

    If set to False, the exception will be printed to console, the flow object will be passed to mitmproxy continuely.

    Note

    When exception occured, the flow object do not always stay intact. This option is only a try-catch like normal Python code. If you run modify1(flow) and modify2(flow) and modify3(flow) and exception raised in modify2(), the flow object will be modified partially.

  • self (str) –

Returns:

The decorated function.

remap_host(flow, overwrite=True)[source]

Remaps the host of the flow to the destination host.

Note

This function is used internally by Xepor. Refer to the source code for customization.

Parameters:

  • flow (HTTPFlow) – The flow to remap.

  • overwrite – Whether to overwrite the host and port of the flow.

Returns:

The remapped host.

get_host(flow)[source]

Gets the host and port of the request. Extending from mitmproxy.http.HTTPFlow.pretty_host to accept values from proxy headers(X-Forwarded-Host and X-Forwarded-Port)

Note

This function is used internally by Xepor. Refer to the source code for customization.

Parameters:

flow (HTTPFlow) – The HTTPFlow object.

Returns:

A tuple of the host and port.

Return type:

Tuple[str, int]

default_response()[source]

This is the default response function for Xepor. It will be called in following conditions:

  1. target host in HTTP request matches the ones in blacklist_domain.

  2. either request_passthrough or response_passthrough is set to False, and no route matches the incoming flow.

Override this function if it suits your needs.

Returns:

A Response object with status code 404 and HTTP header X-Intercepted-By set to xepor.

error_response(msg='APIServer Error')[source]

Returns a response with status code 502 and custom error message.

Override this function if it suits your needs.

Parameters:

msg (str) – The message to be returned.

Returns:

A Response object with status code 502 and content set to the .

find_handler(host, path, rtype=RouteType.REQUEST)[source]

Finds the appropriate handler for the request.

Note

This function is used internally by Xepor. Refer to the source code for customization.

Parameters:

  • host – The host of the request.

  • path – The path of the request.

  • rtype – The type of the route. Accepting RouteType.

Returns:

The handler and the parse result.

class xepor.RouteType(value)[source]

Bases: Enum

This enum is an option set in route definition, specify it to be matched on either incoming request or response.

REQUEST = 1

The route will be matched on mitmproxy request event

RESPONSE = 2

The route will be matched on mitmproxy response event

class xepor.FlowMeta(value)[source]

Bases: Enum

This class is used internally by Xepor to mark flow object by certain metadata. Refer to the source code for detailed usage.

REQ_PASSTHROUGH = 'xepor-request-passthrough'
RESP_PASSTHROUGH = 'xepor-response-passthrough'
REQ_URLPARSE = 'xepor-request-urlparse'
REQ_HOST = 'xepor-request-host'