HowTo: Script Modify Headers for Google Chrome extension using Selenium

Script Modify Headers

		System.setProperty("webdriver.chrome.driver",
				"/opt/tools/selenium-2.45.0/chromedriver");

		File addonpath = new File(
				"/opt/tools/selenium-2.45.0/innpjfdalfhpcoinfnehdnbkglpmogdi.crx");
		ChromeOptions options = new ChromeOptions();
		options.addExtensions(addonpath);
		
		DesiredCapabilities capabilities = new DesiredCapabilities();
		capabilities.setCapability(ChromeOptions.CAPABILITY, options);
		
		WebDriver driver = new ChromeDriver(options);
		driver.get("chrome-extension://innpjfdalfhpcoinfnehdnbkglpmogdi/options.html");
		driver.navigate().refresh();
		Thread.sleep(1000);
		
		driver.findElement(By.xpath("//button[@tooltip='Add New']")).click();
		
		WebElement actionElem = driver.findElement(By.name("action"));
		Select select = new Select(actionElem);
		select.selectByVisibleText("Add");
		
		WebElement nameElem = driver.findElement(By.name("name"));
		nameElem.sendKeys("some-key");

		WebElement valueElem = driver.findElement(By.name("value"));
		valueElem.sendKeys("some-value");

		WebElement descElem = driver.findElement(By.name("description"));
		descElem.sendKeys("some-desc");

		driver.findElement(By.xpath("//button[@tooltip='Save']")).click();
		Thread.sleep(1000);
		
		driver.findElement(By.xpath("//button[@tooltip='Start Modifying Headers']")).click();
		Thread.sleep(1000);
		
		driver.findElement(By.xpath("//button[@tooltip='Enable']")).click();
		Thread.sleep(2000);

		driver.get("http://whatheaders.com");
		Thread.sleep(30000); // Let the user actually see something!
		driver.close();
		driver.quit();

Actually, there's a way without extensions or proxys:

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

def show_image(url:str):  # show image from URL
    from PIL import Image
    from io import BytesIO
    import requests
    try:
        response = requests.get(url)
        img = Image.open(BytesIO(response.content))
        img.show()
    except Exception as e:
        print(e)

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)

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

    return session.listen(devtools.fetch.RequestPaused)

def connection_refused(event, connection):
    print({"type":event.resource_type.to_json(),"frame_id": event.frame_id, "url": event.request.url})

    session, devtools = connection.session, connection.devtools
    # show_image(event.request.url)
    return devtools.fetch.fail_request(request_id=event.request_id,error_reason=devtools.network.ErrorReason.CONNECTION_REFUSED)

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')