Skip to content

pyasic

GoldshellWebAPI

Bases: BaseWebAPI

Source code in pyasic/web/goldshell.py 
 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
class GoldshellWebAPI(BaseWebAPI):
    def __init__(self, ip: str) -> None:
        super().__init__(ip)
        self.username = "admin"
        self.pwd = settings.get("default_goldshell_web_password", "123456789")
        self.token = None

    async def auth(self) -> str | None:
        async with httpx.AsyncClient(transport=settings.transport()) as client:
            try:
                await client.get(f"http://{self.ip}:{self.port}/user/logout")
                auth = (
                    await client.get(
                        f"http://{self.ip}:{self.port}/user/login?username={self.username}&password={self.pwd}&cipher=false"
                    )
                ).json()
            except httpx.HTTPError:
                warnings.warn(f"Could not authenticate web token with miner: {self}")
            except json.JSONDecodeError:
                # try again with encrypted normal password
                try:
                    auth = (
                        await client.get(
                            f"http://{self.ip}:{self.port}/user/login?username=admin&password=bbad7537f4c8b6ea31eea0b3d760e257&cipher=true"
                        )
                    ).json()
                except (httpx.HTTPError, json.JSONDecodeError):
                    warnings.warn(
                        f"Could not authenticate web token with miner: {self}"
                    )
                else:
                    self.token = auth.get("JWT Token")
            else:
                self.token = auth.get("JWT Token")
            return self.token

    async def send_command(
        self,
        command: str | bytes,
        ignore_errors: bool = False,
        allow_warning: bool = True,
        privileged: bool = False,
        **parameters: Any,
    ) -> dict:
        if self.token is None:
            await self.auth()
        async with httpx.AsyncClient(transport=settings.transport()) as client:
            for _ in range(settings.get("get_data_retries", 1)):
                try:
                    if not parameters == {}:
                        response = await client.put(
                            f"http://{self.ip}:{self.port}/mcb/{command}",
                            headers={"Authorization": "Bearer " + self.token},
                            timeout=settings.get("api_function_timeout", 5),
                            json=parameters,
                        )
                    else:
                        response = await client.get(
                            f"http://{self.ip}:{self.port}/mcb/{command}",
                            headers={"Authorization": "Bearer " + self.token},
                            timeout=settings.get("api_function_timeout", 5),
                        )
                    json_data = response.json()
                    return json_data
                except TypeError:
                    await self.auth()
                except (httpx.HTTPError, json.JSONDecodeError):
                    pass

    async def multicommand(
        self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
    ) -> dict:
        data = {k: None for k in commands}
        data["multicommand"] = True
        await self.auth()
        async with httpx.AsyncClient(transport=settings.transport()) as client:
            for command in commands:
                try:
                    response = await client.get(
                        f"http://{self.ip}:{self.port}/mcb/{command}",
                        headers={"Authorization": "Bearer " + self.token},
                        timeout=settings.get("api_function_timeout", 5),
                    )
                    json_data = response.json()
                    data[command] = json_data
                except httpx.HTTPError:
                    pass
                except json.JSONDecodeError:
                    pass
                except TypeError:
                    await self.auth()
        return data

    async def pools(self) -> dict:
        return await self.send_command("pools")

    async def newpool(self, url: str, user: str, password: str) -> dict:
        # looks dumb, but cant pass `pass` since it is a built in type
        return await self.send_command(
            "newpool", **{"url": url, "user": user, "pass": password}
        )

    async def delpool(
        self, url: str, user: str, password: str, dragid: int = 0
    ) -> dict:
        # looks dumb, but cant pass `pass` since it is a built in type
        return await self.send_command(
            "delpool", **{"url": url, "user": user, "pass": password, "dragid": dragid}
        )

    async def setting(self) -> dict:
        return await self.send_command("setting")

    async def set_setting(self, values: dict):
        await self.send_command("setting", **values)

    async def status(self) -> dict:
        return await self.send_command("status")