wkeeling / selenium-wire

Extends Selenium's Python bindings to give you the ability to inspect requests made by the browser.
MIT License
1.86k stars 241 forks source link

[feature request] Change Headers without Proxy or Extension #637

Closed kaliiiiiiiiii closed 1 year ago

kaliiiiiiiiii commented 1 year ago

Actually, I found a quite sneaky way without using a Proxy or Extension:

You can change the Headers in continue_request() at my_headers = {"sec-ch-ua-platform": "Android"} .

# resource: https://stackoverflow.com/questions/66227508/selenium-4-0-0-beta-1-how-add-event-listeners-in-cdp

class cdp_listener:
    from typing import Dict

    def __init__(self):
        self.listeners = {}
    async def async_helper(self):
        async with driver.bidi_connection() as connection:
            session, devtools = connection.session, connection.devtools

            for listener in self.listeners.items():
                my_listener = await listener[1]["listener"](connection=connection)

                async for event in my_listener:
                    try:
                        await session.execute(listener[1]["at_event"](event=event, connection=connection))
                    except Exception as e:
                        print(e)

    def trio_helper(self):
        import trio
        trio.run(self.async_helper)

    def start_threaded(self, listeners: Dict[str,Dict[callable, callable]] = {}):
        if listeners:
            self.listeners = listeners

        import threading
        thread = threading.Thread(target=self.trio_helper)
        thread.start()
        return thread

    def add_listeners(self, listeners: Dict[str,Dict[callable, callable]]):
        self.listeners = listeners

    def remove_listener(self, listener:str):
        del self.listeners[listener]

async def all_requests(connection):
    session, devtools = connection.session, connection.devtools
    pattern = map(devtools.fetch.RequestPattern.from_json,[{"urlPattern":"*"}])
    pattern = list(pattern)
    await session.execute(devtools.fetch.enable(patterns=pattern))

    return session.listen(devtools.fetch.RequestPaused)

def continue_request(event, connection):
    print({"type":event.resource_type.to_json(),"frame_id": event.frame_id, "url": event.request.url})
    session, devtools = connection.session, connection.devtools

    headers = event.request.headers.to_json()

    my_headers = {"sec-ch-ua-platform": "Android"}
    headers.update(my_headers)
    my_headers = []
    for item in headers.items():
        my_headers.append(devtools.fetch.HeaderEntry.from_json({"name": item[0], "value": item[1]}))

    return devtools.fetch.continue_request(request_id=event.request_id, headers=my_headers)

cdp_listener = cdp_listener()
thread = cdp_listener.start_threaded(listeners= {"continue":{"listener":all_requests,"at_event":continue_request}})

driver.get('https://modheader.com/headers?product=ModHeader')
kaliiiiiiiiii commented 1 year ago

resolved