forked from alandtse/auth_capture_proxy
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathauth_capture_proxy.py
More file actions
1010 lines (927 loc) · 43.1 KB
/
Copy pathauth_capture_proxy.py
File metadata and controls
1010 lines (927 loc) · 43.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# SPDX-License-Identifier: Apache-2.0
"""Python Package for auth capture proxy."""
import asyncio
import logging
import posixpath
import re
from functools import partial
from json import JSONDecodeError
from ssl import SSLContext, create_default_context
from typing import Any, Callable, Dict, List, Optional, Set, Text, Tuple, Union
import httpx
from aiohttp import MultipartReader, MultipartWriter, hdrs, web
from multidict import CIMultiDict
from yarl import URL
from authcaptureproxy.const import SKIP_AUTO_HEADERS
from authcaptureproxy.examples.modifiers import (
prepend_relative_urls,
replace_empty_action_urls,
replace_matching_urls,
)
from authcaptureproxy.helper import (
convert_multidict_to_dict,
get_content_type,
get_nested_dict_keys,
print_resp,
run_func,
swap_url,
)
from authcaptureproxy.interceptor import BaseInterceptor, InterceptContext
from authcaptureproxy.stackoverflow import get_open_port
# Pre-configure SSL context
ssl_context = create_default_context()
_LOGGER = logging.getLogger(__name__)
class AuthCaptureProxy:
"""Class to handle proxy login connections.
This class relies on tests to be provided to indicate the proxy has completed. At proxy completion all data can be found in self.session, self.data, and self.query.
"""
def __init__(
self,
proxy_url: URL,
host_url: URL,
session: Optional[httpx.AsyncClient] = None,
session_factory: Optional[Callable[[], httpx.AsyncClient]] = None,
preserve_headers: bool = False,
) -> None:
"""Initialize proxy object.
Args:
proxy_url (URL): url for proxy location. e.g., http://192.168.1.1/. If there is any path, the path is considered part of the base url. If no explicit port is specified, a random port will be generated. If https is passed in, ssl_context must be provided at start_proxy() or the url will be downgraded to http.
host_url (URL): original url for login, e.g., http://example.com
session (httpx.AsyncClient): httpx client to make queries. Optional
session_factory (lambda: httpx.AsyncClient): factory to create the aforementioned httpx client if having one fixed session is insufficient.
preserve_headers (bool): Whether to preserve headers from the backend. Useful in circumventing CSRF protection. Defaults to False.
"""
self._preserve_headers = preserve_headers
self.session_factory: Callable[[], httpx.AsyncClient] = session_factory or (
lambda: httpx.AsyncClient(verify=ssl_context)
)
# NOTE: Do not instantiate httpx.AsyncClient inside the event loop.
# Some SSL initialization (e.g., load_verify_locations) is blocking and will be flagged.
#
# Keep historical behavior when NOT running inside an event loop: create a session immediately.
# When running inside an event loop, defer and create lazily
# via _ensure_session() using asyncio.to_thread().
try:
asyncio.get_running_loop()
in_event_loop = True
except RuntimeError:
in_event_loop = False
if session is not None:
self.session: Optional[httpx.AsyncClient] = session
elif in_event_loop:
self.session = None
else:
self.session = self.session_factory()
self._session_lock = asyncio.Lock()
self._proxy_url: URL = proxy_url
self._host_url: URL = host_url
self._port: int = proxy_url.explicit_port if proxy_url.explicit_port else 0 # type: ignore
self.runner: Optional[web.AppRunner] = None
self.last_resp: Optional[httpx.Response] = None
self.init_query: Dict[Text, Any] = {}
self.query: Dict[Text, Any] = {}
self.data: Dict[Text, Any] = {}
# tests and modifiers should be initialized after port is actually assigned and not during init.
# however, to ensure defaults go first, they should have a dummy key set
self._tests: Dict[Text, Callable] = {}
self._modifiers: Dict[Text, Union[Callable, Dict[Text, Callable]]] = {
"text/html": {
"prepend_relative_urls": lambda x: x,
"change_host_to_proxy": lambda x: x,
}
}
self._old_tests: Dict[Text, Callable] = {}
self._old_modifiers: Dict[Text, Union[Callable, Dict[Text, Callable]]] = {}
self._active = False
self._all_handler_active = True
self.headers: Dict[Text, Text] = {}
self.redirect_filters: Dict[Text, List[Text]] = {
"url": []
} # dictionary of lists of regex strings to filter against
self._background_tasks: Set[asyncio.Task] = set()
self._interceptors: List[BaseInterceptor] = []
@property
def interceptors(self) -> List[BaseInterceptor]:
"""Return interceptors list.
:setter: value (List[BaseInterceptor]): A list of interceptors to run during request processing. See :mod:`authcaptureproxy.examples.amazon_waf` for an example.
"""
return self._interceptors
@interceptors.setter
def interceptors(self, value: List[BaseInterceptor]) -> None:
"""Set interceptors.
Args:
value (List[BaseInterceptor]): A list of interceptors.
"""
self._interceptors = value
@property
def active(self) -> bool:
"""Return whether proxy is started."""
return self._active
@property
def all_handler_active(self) -> bool:
"""Return whether all handler is active."""
return self._all_handler_active
@all_handler_active.setter
def all_handler_active(self, value: bool) -> None:
"""Set all handler to value."""
self._all_handler_active = value
@property
def port(self) -> int:
"""Return port setting."""
return self._port
@property
def tests(self) -> Dict[Text, Callable]:
"""Return tests setting.
:setter: value (Dict[Text, Any]): A dictionary of tests. The key should be the name of the test and the value should be a function or coroutine that takes a httpx.Response, a dictionary of post variables, and a dictioary of query variables and returns a URL or string. See :mod:`authcaptureproxy.examples.testers` for examples.
"""
return self._tests
@tests.setter
def tests(self, value: Dict[Text, Callable]) -> None:
"""Set tests.
Args:
value (Dict[Text, Any]): A dictionary of tests.
"""
self.refresh_tests() # refresh in case of pending change
self._old_tests = self._tests.copy()
self._tests = value
@property
def modifiers(self) -> Dict[Text, Union[Callable, Dict[Text, Callable]]]:
"""Return modifiers setting.
:setter: value (Dict[Text, Dict[Text, Callable]): A nested dictionary of modifiers. The key should be a MIME type and the value should be a dictionary of modifiers for that MIME type where the key should be the name of the modifier and the value should be a function or coroutine that takes a string and returns a modified string. If parameters are necessary, functools.partial should be used. See :mod:`authcaptureproxy.examples.modifiers` for examples.
"""
return self._modifiers
@modifiers.setter
def modifiers(self, value: Dict[Text, Union[Callable, Dict[Text, Callable]]]) -> None:
"""Set tests.
Args:
value (Dict[Text, Any]): A dictionary of tests.
"""
self.refresh_modifiers() # refresh in case of pending change
self._old_modifiers = self._modifiers
self._modifiers = value
def access_url(self) -> URL:
"""Return access url for proxy with port."""
return self._proxy_url.with_port(self.port) if self.port != 0 else self._proxy_url
async def change_host_url(self, new_url: URL) -> None:
"""Change the host url of the proxy.
This will also reset all stored data.
Args:
new_url (URL): original url for login, e.g., http://example.com
"""
if not isinstance(new_url, URL):
raise ValueError("URL required")
self._host_url = new_url
await self.reset_data()
async def reset_data(self) -> None:
"""Reset all stored data.
A proxy may need to service multiple login requests if the route is not torn down. This function will reset all data between logins.
"""
if self.session:
await self.session.aclose()
self.session = None
# Reset data fields unconditionally so state is clean regardless of session outcome.
self.last_resp = None
self.init_query = {}
self.query = {}
self.data = {}
self._active = False
self._all_handler_active = True
await self._ensure_session()
if self.session is None: # pragma: no cover
_LOGGER.error("Internal error: HTTP session not initialized")
return
_LOGGER.debug("Proxy data reset.")
async def _ensure_session(self) -> None:
"""Ensure an httpx session exists.
httpx.AsyncClient() initialization may perform blocking SSL work
(e.g. SSLContext.load_verify_locations), so the client is created
in a background thread.
"""
if self.session is not None:
return
async with self._session_lock:
if self.session is None:
self.session = await asyncio.to_thread(self.session_factory)
def refresh_tests(self) -> None:
"""Refresh tests.
Because tests may use partials, they will freeze their parameters which is a problem with self.access() if the port hasn't been assigned.
"""
if self._tests != self._old_tests:
self.tests.update({})
self.old_tests = self.tests.copy()
_LOGGER.debug("Refreshed %s tests: %s", len(self.tests), list(self.tests.keys()))
def refresh_modifiers(self, site: Optional[URL] = None) -> None:
"""Refresh modifiers.
Because modifiers may use partials, they will freeze their parameters which is a problem with self.access() if the port hasn't been assigned.
Args:
site (Optional[URL], optional): The current site. Defaults to None.
"""
DEFAULT_MODIFIERS = { # noqa: N806
"prepend_relative_urls": partial(prepend_relative_urls, self.access_url()),
"change_host_to_proxy": partial(
replace_matching_urls,
self._host_url.with_query({}).with_path("/"),
self.access_url(),
),
}
if self._modifiers != self._old_modifiers:
if self.modifiers.get("text/html") is None:
self.modifiers["text/html"] = DEFAULT_MODIFIERS # type: ignore
elif self.modifiers.get("text/html") and isinstance(self.modifiers["text/html"], dict):
self.modifiers["text/html"].update(DEFAULT_MODIFIERS)
if site and isinstance(self.modifiers["text/html"], dict):
self.modifiers["text/html"].update(
{
"change_empty_to_proxy": partial(
replace_empty_action_urls,
swap_url(
old_url=self._host_url.with_query({}),
new_url=self.access_url().with_query({}),
url=site,
),
),
}
)
self._old_modifiers = self.modifiers.copy()
refreshed_modifers = get_nested_dict_keys(self.modifiers)
_LOGGER.debug("Refreshed %s modifiers: %s", len(refreshed_modifers), refreshed_modifers)
@staticmethod
def _filter_ajax_headers(resp: httpx.Response) -> dict:
"""Filter headers for AJAX responses, removing hop-by-hop and CSP headers."""
_skip_headers = {
"content-type",
"content-length",
"content-encoding",
"transfer-encoding",
"connection",
"x-connection-hash",
"content-security-policy",
"content-security-policy-report-only",
}
filtered = {}
for k, v in resp.headers.items():
if k.lower() not in _skip_headers:
filtered[k] = v
filtered["Cache-Control"] = "no-cache, no-store, must-revalidate"
return filtered
async def _build_response(
self, response: Optional[httpx.Response] = None, *args, **kwargs
) -> web.Response:
"""
Build a response.
"""
if "headers" not in kwargs and response is not None:
kwargs["headers"] = response.headers.copy() if self._preserve_headers else CIMultiDict()
if hdrs.CONTENT_TYPE in kwargs["headers"] and "content_type" in kwargs:
del kwargs["headers"][hdrs.CONTENT_TYPE]
if hdrs.CONTENT_LENGTH in kwargs["headers"]:
del kwargs["headers"][hdrs.CONTENT_LENGTH]
if hdrs.CONTENT_ENCODING in kwargs["headers"]:
del kwargs["headers"][hdrs.CONTENT_ENCODING]
if hdrs.CONTENT_TRANSFER_ENCODING in kwargs["headers"]:
del kwargs["headers"][hdrs.CONTENT_TRANSFER_ENCODING]
if hdrs.TRANSFER_ENCODING in kwargs["headers"]:
del kwargs["headers"][hdrs.TRANSFER_ENCODING]
if "x-connection-hash" in kwargs["headers"]:
del kwargs["headers"]["x-connection-hash"]
while hdrs.SET_COOKIE in kwargs["headers"]:
del kwargs["headers"][hdrs.SET_COOKIE]
# cache control
if hdrs.CACHE_CONTROL in kwargs["headers"]:
del kwargs["headers"][hdrs.CACHE_CONTROL]
kwargs["headers"][hdrs.CACHE_CONTROL] = "no-cache, no-store, must-revalidate"
return web.Response(*args, **kwargs)
async def all_handler(self, request: web.Request, **kwargs) -> web.Response:
"""Handle all requests.
This handler will exit on successful test found in self.tests or if a /stop url is seen. This handler can be used with any aiohttp webserver and disabled after registered using self.all_haandler_active.
The handler supports an interceptor pipeline for extending behavior
without modifying core proxy code. See :class:`BaseInterceptor`.
Args
request (web.Request): The request to process
**kwargs: Additional keyword arguments
access_url (URL): The access url for the proxy. Defaults to self.access_url()
host_url (URL): The host url for the proxy. Defaults to self._host_url
Returns
web.Response: The webresponse to the browser
Raises
web.HTTPFound: Redirect URL upon success
web.HTTPNotFound: Return 404 when all_handler is disabled
"""
if "access_url" in kwargs:
access_url = kwargs.pop("access_url")
else:
access_url = self.access_url()
if "host_url" in kwargs:
host_url = kwargs.pop("host_url")
else:
host_url = self._host_url
# Ensure the HTTP session is created off the event loop thread.
await self._ensure_session()
session = self.session
if session is None: # pragma: no cover
return await self._build_response(
text="Internal error: HTTP session not initialized", status=500
)
async def _process_multipart(reader: MultipartReader, writer: MultipartWriter) -> None:
"""Process multipart.
Args:
reader (MultipartReader): Response multipart to process.
writer (MultipartWriter): Multipart to write out.
"""
while True:
part = await reader.next() # noqa: B305
# https://github.com/PyCQA/flake8-bugbear/issues/59
if part is None:
break
if isinstance(part, MultipartReader):
await _process_multipart(part, writer)
elif hdrs.CONTENT_TYPE in part.headers:
content_type = part.headers.get(hdrs.CONTENT_TYPE, "")
mime_type = content_type.split(";", 1)[0].strip()
if mime_type == "application/json":
try:
part_data: Optional[
Union[Text, Dict[Text, Any], List[Tuple[Text, Text]], bytes]
] = await part.json()
writer.append_json(part_data)
except (JSONDecodeError, ValueError, TypeError):
# Best-effort fallback: text, then bytes
try:
part_text = await part.text()
writer.append(part_text)
except ValueError:
part_data = await part.read()
writer.append(part_data)
elif mime_type.startswith("text"):
part_data = await part.text()
writer.append(part_data)
elif mime_type == "application/x-www-form-urlencoded":
part_data = await part.form()
writer.append_form(part_data)
else:
part_data = await part.read()
writer.append(part_data)
else:
part_data = await part.read()
if part.name:
self.data.update({part.name: part_data})
elif part.filename:
part_data = await part.read()
self.data.update({part.filename: part_data})
writer.append(part_data)
if not self.all_handler_active:
_LOGGER.debug("%s all_handler is disabled; returning 404.", self)
raise web.HTTPNotFound()
method = request.method.lower()
_LOGGER.debug("Received %s: %s for %s", method, str(request.url), host_url)
resp: Optional[httpx.Response] = None
# Create interceptor context for the request pipeline
ctx = InterceptContext(
request=request,
proxy=self,
access_url=access_url,
host_url=host_url,
method=method,
)
# Run on_request interceptors (can set ctx.site for custom URL routing)
for interceptor in self._interceptors:
await interceptor.on_request(ctx)
if ctx.short_circuit is not None:
return ctx.short_circuit
if ctx.site:
# Interceptor set the target URL (e.g., multi-host routing)
site = ctx.site
else:
# Generic URL resolution
old_url: URL = (
access_url.with_host(request.url.host)
if request.url.host and request.url.host != access_url.host
else access_url
)
if request.scheme == "http" and access_url.scheme == "https":
_LOGGER.debug("Detected http while should be https; switching to https")
site = str(
swap_url(
ignore_query=True,
old_url=old_url.with_scheme("https"),
new_url=host_url.with_path("/"),
url=URL(str(request.url)).with_scheme("https"),
),
)
else:
site = str(
swap_url(
ignore_query=True,
old_url=old_url,
new_url=host_url.with_path("/"),
url=URL(str(request.url)),
),
)
self.query.update(request.query)
data: Optional[Dict] = None
raw_body: Optional[bytes] = None
mpwriter = None
if request.content_type == "multipart/form-data":
mpwriter = MultipartWriter()
await _process_multipart(await request.multipart(), mpwriter)
elif (
request.has_body
and request.content_type
and "x-www-form-urlencoded" not in request.content_type
and "json" not in request.content_type
):
# Raw body (text/plain, binary, etc.) - forward as-is.
raw_body = await request.read()
_LOGGER.debug(
"Read raw body (%s bytes, type=%s) for %s",
len(raw_body) if raw_body else 0,
request.content_type,
site,
)
else:
data = convert_multidict_to_dict(await request.post())
json_data = None
# Only attempt JSON decoding for JSON requests; avoid raising for form posts.
if request.has_body and (
request.content_type == "application/json" or request.content_type.endswith("+json")
):
try:
json_data = await request.json()
except (JSONDecodeError, ValueError):
json_data = None
if data:
self.data.update(data)
_LOGGER.debug("Storing data %s", data)
# Run on_request_data interceptors (can modify data before HTTP request)
ctx.site = site
ctx.data = data
ctx.json_data = json_data
for interceptor in self._interceptors:
await interceptor.on_request_data(ctx)
if ctx.short_circuit is not None:
return ctx.short_circuit
data = ctx.data
elif json_data:
self.data.update(json_data)
_LOGGER.debug("Storing json %s", json_data)
if URL(str(request.url)).path == re.sub(
r"/+", "/", self._proxy_url.with_path(f"{self._proxy_url.path}/stop").path
):
self.all_handler_active = False
if self.active:
task = asyncio.create_task(self.stop_proxy(3))
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
return await self._build_response(text="Proxy stopped.")
elif (
URL(str(request.url)).path
== re.sub(r"/+", "/", self._proxy_url.with_path(f"{self._proxy_url.path}/resume").path)
and self.last_resp
and isinstance(self.last_resp, httpx.Response)
):
self.init_query = self.query.copy()
_LOGGER.debug("Resuming request: %s", self.last_resp)
resp = self.last_resp
else:
if URL(str(request.url)).path in [
self._proxy_url.path,
re.sub(
r"/+", "/", self._proxy_url.with_path(f"{self._proxy_url.path}/resume").path
),
]:
# either base path or resume without anything to resume
site = str(URL(host_url))
if method == "get":
self.init_query = self.query.copy()
_LOGGER.debug(
"Starting auth capture proxy for %s",
host_url,
)
headers = await self.modify_headers(URL(site), request)
skip_auto_headers: List[str] = headers.get(SKIP_AUTO_HEADERS, [])
if skip_auto_headers:
_LOGGER.debug("Discovered skip_auto_headers %s", skip_auto_headers)
headers.pop(SKIP_AUTO_HEADERS)
# Avoid accidental header mutation across branches/calls
req_headers: dict[str, Any] = dict(headers)
_LOGGER.debug(
"Attempting %s to %s\nheaders: %s \ncookies: %s",
method,
site,
req_headers,
session.cookies.jar,
)
try:
if mpwriter:
resp = await getattr(session, method)(
site, data=mpwriter, headers=req_headers, follow_redirects=True
)
elif data:
resp = await getattr(session, method)(
site, data=data, headers=req_headers, follow_redirects=True
)
elif raw_body is not None:
_LOGGER.debug(
"Sending raw body (%s bytes, Content-Type: %s) to %s",
len(raw_body),
request.content_type,
site,
)
# Preserve the original Content-Type for raw body requests
if request.content_type and "Content-Type" not in req_headers:
req_headers["Content-Type"] = request.content_type
resp = await getattr(session, method)(
site, content=raw_body, headers=req_headers, follow_redirects=True
)
elif json_data:
for item in ["Host", "Origin", "User-Agent", "dnt", "Accept-Encoding"]:
# remove proxy headers
if req_headers.get(item):
req_headers.pop(item)
resp = await getattr(session, method)(
site, json=json_data, headers=req_headers, follow_redirects=True
)
else:
resp = await getattr(session, method)(
site, headers=req_headers, follow_redirects=True
)
except httpx.ConnectError as ex:
return await self._build_response(
text=f"Error connecting to {site}; please retry: {ex}"
)
except httpx.TooManyRedirects as ex:
return await self._build_response(
text=f"Error connecting to {site}; too many redirects: {ex}"
)
except httpx.TimeoutException as ex:
_LOGGER.warning("Timeout connecting to %s: %s", site, ex)
return await self._build_response(
text=(
f"Timeout connecting to {site}: {ex}. "
"Please try again. If this persists, check your network "
"and that the service endpoint is reachable from this host."
)
)
except httpx.HTTPError as ex:
return await self._build_response(text=f"Error connecting to {site}: {ex}")
if resp is None:
return await self._build_response(text=f"Error connecting to {site}; please retry")
self.last_resp = resp
print_resp(resp)
# Run on_response interceptors (post-response, pre-tests)
ctx.response = resp
ctx.site = site
for interceptor in self._interceptors:
await interceptor.on_response(ctx)
if ctx.short_circuit is not None:
return ctx.short_circuit
self.check_redirects()
self.refresh_tests()
if self.tests:
for test_name, test in self.tests.items():
result = None
result = await run_func(test, test_name, resp, self.data, self.query)
if result:
_LOGGER.debug("Test %s triggered", test_name)
if isinstance(result, URL):
_LOGGER.debug(
"Redirecting to callback: %s",
result,
)
raise web.HTTPFound(location=result)
elif isinstance(result, str):
_LOGGER.debug("Displaying page:\n%s", result)
return await self._build_response(
resp, text=result, content_type="text/html"
)
else:
_LOGGER.warning("Proxy has no tests; please set.")
content_type = get_content_type(resp)
# Detect AJAX requests using Fetch Metadata headers (W3C standard).
# Sec-Fetch-Mode is set by the browser and cannot be spoofed by JS.
# 'navigate' = top-level page navigation; anything else = AJAX/subresource.
# Fall back to Upgrade-Insecure-Requests for older clients.
_sec_fetch_mode = request.headers.get("Sec-Fetch-Mode")
if _sec_fetch_mode is not None:
_is_ajax = _sec_fetch_mode != "navigate"
else:
# Legacy fallback for clients without Sec-Fetch-Mode
_is_ajax = request.headers.get("Upgrade-Insecure-Requests") != "1"
if _is_ajax:
_LOGGER.debug(
"AJAX response for %s: status=%s, content_type=%s",
URL(str(request.url)).path,
resp.status_code,
content_type,
)
if _is_ajax and content_type == "text/html":
_ajax_body = resp.content
# Run on_ajax_html interceptors (can modify AJAX HTML body)
ctx.is_ajax = True
ctx.content_type = content_type
ctx.body = _ajax_body
for interceptor in self._interceptors:
await interceptor.on_ajax_html(ctx)
_ajax_body = ctx.body if ctx.body is not None else _ajax_body
_LOGGER.debug(
"AJAX HTML response for %s - skipping modifiers",
URL(str(request.url)).path,
)
# Forward original headers for AJAX responses.
# Client-side JavaScript may check response headers (e.g., for CAPTCHA
# initialization). Without them, it may fail silently.
_ajax_headers = self._filter_ajax_headers(resp) if resp is not None else {}
return await self._build_response(
resp,
body=_ajax_body,
content_type=content_type,
headers=_ajax_headers,
)
# Also skip modifiers for non-HTML AJAX responses (JSON, binary, etc.)
if _is_ajax and content_type != "text/html":
_LOGGER.debug(
"AJAX non-HTML response (%s) for %s - skipping modifiers",
content_type,
URL(str(request.url)).path,
)
_resp_body = resp.content
_ajax_headers_nh = self._filter_ajax_headers(resp) if resp is not None else {}
return await self._build_response(
resp,
body=_resp_body,
content_type=content_type,
headers=_ajax_headers_nh,
)
self.refresh_modifiers(URL(str(resp.url)))
if self.modifiers:
modified: bool = False
if content_type != "text/html" and content_type not in self.modifiers.keys():
text: Text = ""
elif content_type != "text/html" and content_type in self.modifiers.keys():
text = resp.text
else:
text = resp.text
if not isinstance(text, str): # process aiohttp text
text = await resp.text()
# Resolve relative form actions BEFORE modifiers run.
if text and content_type == "text/html" and resp and resp.url:
_resp_url = URL(str(resp.url))
_resp_dir = _resp_url.path.rsplit("/", 1)[0] + "/" if "/" in _resp_url.path else "/"
def _resolve_form_action(form_match):
"""Resolve relative action URLs only inside <form> tags."""
form_tag = form_match.group(0)
action_m = re.search(r'(\s+action=["\'])([^"\']*?)(["\'])', form_tag)
if not action_m:
return form_tag
action = action_m.group(2)
if action and not action.startswith(
("http://", "https://", "//", "#", "javascript:", "/")
):
resolved_path = posixpath.normpath(_resp_dir + action)
_proxy_base = self.access_url().path.rstrip("/")
abs_url = str(
self.access_url().with_path(_proxy_base + resolved_path).with_query({})
)
_LOGGER.debug(
"Resolved relative form action '%s' -> '%s' (page: %s)",
action,
abs_url,
_resp_url.path,
)
return form_tag[: action_m.start(2)] + abs_url + form_tag[action_m.end(2) :]
return form_tag
text = re.sub(
r"<form\b[^>]*>",
_resolve_form_action,
text,
flags=re.IGNORECASE,
)
# Run on_page_html interceptors (can inject scripts before modifiers)
if text and content_type == "text/html":
ctx.text = text
ctx.content_type = content_type
ctx.is_ajax = False
for interceptor in self._interceptors:
await interceptor.on_page_html(ctx)
text = ctx.text if ctx.text is not None else text
if text:
for name, modifier in self.modifiers.items():
if isinstance(modifier, dict):
if name != content_type:
continue
for sub_name, sub_modifier in modifier.items():
try:
text = await run_func(sub_modifier, sub_name, text)
modified = True
except TypeError as ex:
_LOGGER.warning("Modifier %s is not callable: %s", sub_name, ex)
else:
# default run against text/html only
if content_type == "text/html":
try:
text = await run_func(modifier, name, text)
modified = True
except TypeError as ex:
_LOGGER.warning("Modifier %s is not callable: %s", name, ex)
if modified:
return await self._build_response(
resp,
text=text,
content_type=content_type,
)
# pass through non parsed content
_LOGGER.debug(
"Passing through %s as %s",
(
URL(str(request.url)).name
if URL(str(request.url)).name
else URL(str(request.url)).path
),
content_type,
)
return await self._build_response(resp, body=resp.content, content_type=content_type)
async def start_proxy(
self, host: Optional[Text] = None, ssl_context: Optional[SSLContext] = None
) -> None:
"""Start proxy.
Args:
host (Optional[Text], optional): The host interface to bind to. Defaults to None which is "0.0.0.0" all interfaces.
ssl_context (Optional[SSLContext], optional): SSL Context for the server. Defaults to None.
"""
app = web.Application()
app.add_routes(
[
web.route("*", "/{tail:.*}", self.all_handler),
]
)
self.runner = web.AppRunner(app)
await self.runner.setup()
if not self.port:
self._port = get_open_port()
if self._proxy_url.scheme == "https" and ssl_context is None:
_LOGGER.debug("Proxy url is https but no SSL Context set, downgrading to http")
self._proxy_url = self._proxy_url.with_scheme("http")
site = web.TCPSite(runner=self.runner, host=host, port=self.port, ssl_context=ssl_context)
await site.start()
self._active = True
_LOGGER.debug("Started proxy at %s", self.access_url())
async def stop_proxy(self, delay: int = 0) -> None:
"""Stop proxy server.
Args:
delay (int, optional): How many seconds to delay. Defaults to 0.
"""
if not self.active:
_LOGGER.debug("Proxy is not started; ignoring stop command")
return
_LOGGER.debug("Stopping proxy at %s after %s seconds", self.access_url(), delay)
await asyncio.sleep(delay)
_LOGGER.debug("Closing site runner")
if self.runner:
await self.runner.cleanup()
await self.runner.shutdown()
_LOGGER.debug("Site runner closed")
# close session
if self.session:
_LOGGER.debug("Closing session")
await self.session.aclose()
self.session = None
_LOGGER.debug("Session closed")
self._active = False
_LOGGER.debug("Proxy stopped")
def _swap_proxy_and_host(self, text: Text, domain_only: bool = False) -> Text:
"""Replace host with proxy address or proxy with host address.
Args
text (Text): text to replace
domain (bool): Whether only the domains should be swapped.
Returns
Text: Result of replacing
"""
host_string: Text = str(self._host_url.with_path("/"))
proxy_string: Text = str(
self.access_url() if not domain_only else self.access_url().with_path("/")
)
if str(self.access_url().with_path("/")).replace("https", "http") in text:
_LOGGER.debug(
"Replacing %s with %s",
str(self.access_url().with_path("/")).replace("https", "http"),
str(self.access_url().with_path("/")),
)
text = text.replace(
str(self.access_url().with_path("/")).replace("https", "http"),
str(self.access_url().with_path("/")),
)
if proxy_string in text:
if host_string[-1] == "/" and (
not proxy_string or proxy_string == "/" or proxy_string[-1] != "/"
):
proxy_string = f"{proxy_string}/"
_LOGGER.debug("Replacing %s with %s in %s", proxy_string, host_string, text)
return text.replace(proxy_string, host_string)
elif host_string in text:
if host_string[-1] == "/" and (
not proxy_string or proxy_string == "/" or proxy_string[-1] != "/"
):
proxy_string = f"{proxy_string}/"
_LOGGER.debug("Replacing %s with %s", host_string, proxy_string)
return text.replace(host_string, proxy_string)
else:
_LOGGER.debug("Unable to find %s and %s in %s", host_string, proxy_string, text)
return text
async def modify_headers(self, site: URL, request: web.Request) -> dict:
"""Modify headers.
Return modified headers based on site and request. To disable auto header generation,
pass in to the header a key const.SKIP_AUTO_HEADERS with a list of keys to not generate.
For example, to prevent User-Agent generation: {SKIP_AUTO_HEADERS : ["User-Agent"]}
Args:
site (URL): URL of the next host request.
request (web.Request): Proxy directed request. This will need to be changed for the actual host request.
Returns:
dict: Headers after modifications
"""
result: Dict[str, Any] = {}
result.update(request.headers)
# _LOGGER.debug("Original headers %s", headers)
if result.get("Host"):
result.pop("Host")
if result.get("Origin"):
# Use the configured host URL as Origin for cross-origin requests.
# Third-party services may validate Origin against the page that loaded them.
result["Origin"] = f"{self._host_url.with_path('')}"
# remove any cookies in header received from browser. If not removed, httpx will not send session cookies
if result.get("Cookie"):
result.pop("Cookie")
if result.get("Referer") and (
URL(result.get("Referer", "")).query == self.init_query
or URL(result.get("Referer", "")).path
== "/config/integrations" # home-assistant referer
):
# Change referer for starting request; this may have query items we shouldn't pass
result["Referer"] = str(self._host_url)
elif result.get("Referer"):
result["Referer"] = self._swap_proxy_and_host(
result.get("Referer", ""), domain_only=True
)
for item in [
"Content-Length",
"X-Forwarded-For",
"X-Forwarded-Host",
"X-Forwarded-Port",
"X-Forwarded-Proto",
"X-Forwarded-Scheme",
"X-Forwarded-Server",
"X-Real-IP",
]:
# remove proxy headers
if result.get(item):
result.pop(item)
result.update(self.headers if self.headers else {})
_LOGGER.debug("Final headers %s", result)
return result
def check_redirects(self) -> None:
"""Change host if redirect detected and regex does not match self.redirect_filters.
Self.redirect_filters is a dict with key as attr in resp and value as list of regex expressions to filter against.
"""
if not self.last_resp:
return
resp: httpx.Response = self.last_resp
if resp.history:
for item in resp.history:
if (
item.status_code in [301, 302, 303, 304, 305, 306, 307, 308]
and item.url
and resp.url
and resp.url.host != self._host_url.host
):
filtered = False
for attr, regex_list in self.redirect_filters.items():
if getattr(resp, attr) and list(
filter(
lambda regex_string: re.search(
regex_string, str(getattr(resp, attr))
),
regex_list,
)
):
_LOGGER.debug(
"Check_redirects: Filtered out on %s in %s for resp attribute %s",
list(
filter(
lambda regex_string: re.search(
regex_string, str(getattr(resp, attr))
),
regex_list,
)
),
str(getattr(resp, attr)),
attr,
)