Decryption-Tools/pywidevine
1
0
Fork 0
mirror of https://github.com/devine-dl/pywidevine.git synced 2024-10-30 05:29:21 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

serve: Set Server response header with pywidevine version

Browse Source 
This allows clients to test with a HEAD request to / to see what version the API is running and test if it's actually a pywidevine serve API.
This commit is contained in:
rlaphoenix 2022-08-02 00:18:08 +01:00
parent 4f96ee402b
commit 4bc0edcca9
1 changed files with 31 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
pywidevine/serve.py
View File

@ -3,8 +3,6 @@ import sys
from pathlib import Path from pathlib import Path
from typing import Optional, Union from typing import Optional, Union
from pywidevine.exceptions import TooManySessions, InvalidSession
try: try:
from aiohttp import web from aiohttp import web
except ImportError: except ImportError:
@ -15,8 +13,10 @@ except ImportError:
) )
sys.exit(1) sys.exit(1)
from pywidevine import __version__
from pywidevine.cdm import Cdm from pywidevine.cdm import Cdm
from pywidevine.device import Device from pywidevine.device import Device
from pywidevine.exceptions import TooManySessions, InvalidSession
from pywidevine.license_protocol_pb2 import LicenseType, License from pywidevine.license_protocol_pb2 import LicenseType, License
routes = web.RouteTableDef() routes = web.RouteTableDef()
@ -256,30 +256,38 @@ async def keys(request: web.Request) -> web.Response:
@web.middleware @web.middleware
async def authentication(request: web.Request, handler) -> web.Response: async def authentication(request: web.Request, handler) -> web.Response:
response = None
if request.path != "/":
secret_key = request.headers.get("X-Secret-Key") secret_key = request.headers.get("X-Secret-Key")
if not secret_key: if not secret_key:
request.app.logger.debug(f"{request.remote} did not provide authorization.") request.app.logger.debug(f"{request.remote} did not provide authorization.")
return web.json_response({ response = web.json_response({
"status": "401", "status": "401",
"message": "Secret Key is Empty." "message": "Secret Key is Empty."
}, status=401) }, status=401)
elif secret_key not in request.app["config"]["users"]:
if secret_key not in request.app["config"]["users"]:
request.app.logger.debug(f"{request.remote} failed authentication with '{secret_key}'.") request.app.logger.debug(f"{request.remote} failed authentication with '{secret_key}'.")
return web.json_response({ response = web.json_response({
"status": "401", "status": "401",
"message": "Secret Key is Invalid, the Key is case-sensitive." "message": "Secret Key is Invalid, the Key is case-sensitive."
}, status=401) }, status=401)
if response is None:
try: try:
return await handler(request) response = await handler(request)
except web.HTTPException as e: except web.HTTPException as e:
request.app.logger.error(f"An unexpected error has occurred, {e}") request.app.logger.error(f"An unexpected error has occurred, {e}")
return web.json_response({ response = web.json_response({
"status": 500, "status": 500,
"message": e.reason "message": e.reason
}, status=500) }, status=500)
response.headers.update({
"Server": f"https://github.com/rlaphoenix/pywidevine serve v{__version__}"
})
return response
def run(config: dict, host: Optional[Union[str, web.HostSequence]] = None, port: Optional[int] = None): def run(config: dict, host: Optional[Union[str, web.HostSequence]] = None, port: Optional[int] = None):
app = web.Application(middlewares=[authentication]) app = web.Application(middlewares=[authentication])