Devine
/
pywidevine
mirror of https://github.com/devine-dl/pywidevine.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RemoteCdm: Improve API error handling

This commit is contained in:
rlaphoenix 2022-08-06 10:12:11 +01:00
parent 2179987986
commit 2e2b5d528a
1 changed files with 22 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
pywidevine/remotecdm.py
View File

@ -94,10 +94,12 @@ class RemoteCdm(Cdm):
raise NotImplementedError("You cannot load a RemoteCdm from a local Device file.") raise NotImplementedError("You cannot load a RemoteCdm from a local Device file.")
def open(self) -> bytes: def open(self) -> bytes:
r = self.__session.get(f"{self.host}/{self.device_name}/open") r = self.__session.get(
if r.status_code != 200: url=f"{self.host}/{self.device_name}/open"
raise ValueError(f"Cannot Open CDM Session, {r.text} [{r.status_code}]") ).json()
r = r.json()["data"] if r['status'] != 200:
raise ValueError(f"Cannot Open CDM Session, {r['message']} [{r['status']}]")
r = r["data"]
if int(r["device"]["system_id"]) != self.system_id: if int(r["device"]["system_id"]) != self.system_id:
raise DeviceMismatch("The System ID specified does not match the one specified in the API response.") raise DeviceMismatch("The System ID specified does not match the one specified in the API response.")
@ -108,9 +110,11 @@ class RemoteCdm(Cdm):
return bytes.fromhex(r["session_id"]) return bytes.fromhex(r["session_id"])
def close(self, session_id: bytes) -> None: def close(self, session_id: bytes) -> None:
r = self.__session.get(f"{self.host}/{self.device_name}/close/{session_id.hex()}") r = self.__session.get(
if r.status_code != 200: url=f"{self.host}/{self.device_name}/close/{session_id.hex()}"
raise ValueError(f"Cannot Close CDM Session, {r.text} [{r.status_code}]") ).json()
if r["status"] != 200:
raise ValueError(f"Cannot Close CDM Session, {r['message']} [{r['status']}]")
def set_service_certificate(self, session_id: bytes, certificate: Optional[Union[bytes, str]]) -> str: def set_service_certificate(self, session_id: bytes, certificate: Optional[Union[bytes, str]]) -> str:
if certificate is None: if certificate is None:
@ -128,10 +132,10 @@ class RemoteCdm(Cdm):
"session_id": session_id.hex(), "session_id": session_id.hex(),
"certificate": certificate_b64 "certificate": certificate_b64
} }
) ).json()
if r.status_code != 200: if r["status"] != 200:
raise ValueError(f"Cannot Set CDMs Service Certificate, {r.text} [{r.status_code}]") raise ValueError(f"Cannot Set CDMs Service Certificate, {r['message']} [{r['status']}]")
r = r.json()["data"] r = r["data"]
return r["provider_id"] return r["provider_id"]
@ -165,10 +169,10 @@ class RemoteCdm(Cdm):
"session_id": session_id.hex(), "session_id": session_id.hex(),
"init_data": pssh.dumps() "init_data": pssh.dumps()
} }
) ).json()
if r.status_code != 200: if r["status"] != 200:
raise ValueError(f"Cannot get Challenge, {r.text} [{r.status_code}]") raise ValueError(f"Cannot get Challenge, {r['message']} [{r['status']}]")
r = r.json()["data"] r = r["data"]
try: try:
license_message = SignedMessage() license_message = SignedMessage()
@ -211,9 +215,9 @@ class RemoteCdm(Cdm):
"session_id": session_id.hex(), "session_id": session_id.hex(),
"license_message": base64.b64encode(license_message.SerializeToString()).decode() "license_message": base64.b64encode(license_message.SerializeToString()).decode()
} }
) ).json()
if r.status_code != 200: if r["status"] != 200:
raise ValueError(f"Cannot parse License, {r.text} [{r.status_code}]") raise ValueError(f"Cannot parse License, {r['message']} [{r['status']}]")
def get_keys(self, session_id: bytes, type_: Optional[Union[int, str]] = None) -> list[Key]: def get_keys(self, session_id: bytes, type_: Optional[Union[int, str]] = None) -> list[Key]:
try: try: