diff --git a/axis/__main__.py b/axis/__main__.py index 3aa0cb96..617a2643 100644 --- a/axis/__main__.py +++ b/axis/__main__.py @@ -20,8 +20,13 @@ def event_handler(event: Event) -> None: async def axis_device( - host: str, port: int, username: str, password: str, is_companion: bool = False -) -> AxisDevice: + host: str, + port: int, + username: str, + password: str, + web_proto: str, + is_companion: bool = False, +) -> axis.device.AxisDevice: """Create a Axis device.""" session = AsyncClient(verify=False) # noqa: S501 device = AxisDevice( @@ -32,6 +37,7 @@ async def axis_device( username=username, password=password, is_companion=is_companion, + web_proto=web_proto, ) ) @@ -58,12 +64,18 @@ async def axis_device( async def main( - host: str, port: int, username: str, password: str, params: bool, events: bool + host: str, + port: int, + username: str, + password: str, + params: bool, + events: bool, + web_proto: str, ) -> None: """CLI method for library.""" LOGGER.info("Connecting to Axis device") - device = await axis_device(host, port, username, password) + device = await axis_device(host, port, username, password, web_proto=web_proto) if not device: LOGGER.error("Couldn't connect to Axis device") @@ -96,6 +108,7 @@ async def main( parser.add_argument("username", type=str) parser.add_argument("password", type=str) parser.add_argument("-p", "--port", type=int, default=80) + parser.add_argument("--proto", type=str, default="http") parser.add_argument("--events", action="store_true") parser.add_argument("--params", action="store_true") parser.add_argument("-D", "--debug", action="store_true") @@ -125,6 +138,7 @@ async def main( port=args.port, params=args.params, events=args.events, + web_proto=args.proto, ) ) diff --git a/axis/interfaces/vapix.py b/axis/interfaces/vapix.py index 63c39aef..89a6ae2e 100644 --- a/axis/interfaces/vapix.py +++ b/axis/interfaces/vapix.py @@ -42,12 +42,19 @@ TIME_OUT = 15 +class WrongAuthError(RequestError): + """Wrong authentication method response.""" + + class Vapix: """Vapix parameter request.""" + auth: httpx.Auth + def __init__(self, device: AxisDevice) -> None: """Store local reference to device config.""" self.device = device + # self.auth = httpx.BasicAuth(device.config.username, device.config.password) self.auth = httpx.DigestAuth(device.config.username, device.config.password) self.users = Users(self) @@ -256,6 +263,35 @@ async def request( data: dict[str, str] | None = None, headers: dict[str, str] | None = None, params: dict[str, str] | None = None, + ) -> bytes: + """Make a request to the device.""" + try: + return await self._request( + method=method, + path=path, + content=content, + data=data, + headers=headers, + params=params, + ) + except WrongAuthError: + return await self._request( + method=method, + path=path, + content=content, + data=data, + headers=headers, + params=params, + ) + + async def _request( + self, + method: str, + path: str, + content: bytes | None = None, + data: dict[str, str] | None = None, + headers: dict[str, str] | None = None, + params: dict[str, str] | None = None, ) -> bytes: """Make a request to the device.""" url = self.device.config.url + path @@ -291,6 +327,7 @@ async def request( response.raise_for_status() except httpx.HTTPStatusError as errh: + self._evaluate_auth(response.headers) LOGGER.debug("%s, %s", response, errh) raise_error(response.status_code) @@ -302,3 +339,20 @@ async def request( ) return response.content + + def _evaluate_auth(self, headers: httpx.Headers) -> None: + """Evaluate and reassign authentication method if needed.""" + dev = self.device + expected_auth = headers.get("www-authenticate", "").lower() + + auth_mapping = { + "basic": (httpx.DigestAuth, httpx.BasicAuth), + "digest": (httpx.BasicAuth, httpx.DigestAuth), + } + + for auth_type, (current_auth, new_auth) in auth_mapping.items(): + if expected_auth.startswith(auth_type) and isinstance( + self.auth, current_auth + ): + self.auth = new_auth(dev.config.username, dev.config.password) + raise WrongAuthError