Freebox thin client
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

156 lines
5.1 KiB

Fbx Client
import os
import logging
import re
import hmac
import hashlib
import logging
import time
import requests
from . import api
from . import utils
from . import mdns
__all__ = [
"FbxClass", ]
class FbxClass():
Base class for Fbx subsystems
def __init__(self, transport):
self._trn = transport
class FbxTransport():
Transport abstraction and context handling for all methods
def __init__(self, url=None, session=None):
self._session = session or requests.session()
self._session.verify = os.path.join(os.path.dirname(__file__), 'fb.pem')
self.log = logging.getLogger("pyfbx.trans")
def set_url(self, url):
if url is None:
self._url = mdns.FbxMDNS().search() or self.get_local_base()
if"https?://", url) is None:
url = "http://" + url
if "/api/" not in url:
self._url = self.get_local_base(url)
self._url = url
def set_session_header(self, session_token):
self._session.headers.update({'X-Fbx-App-Auth': session_token})
def api_exec(self, http_method, endpoint, post_data=None, **kwargs):
req_response = self._session.request(
http_method, self._url + "/" + endpoint.format(**kwargs), json=post_data)
self.log.debug(">> Sent %s %s/%s Post: %s", http_method, self._url,
endpoint.format(**kwargs), post_data)
response = req_response.json()
if response['success']:
if 'result' in response:
self.log.debug(f"<< Got {response['result']}")
return response['result']
raise FbxErrorResponse(response['error_code'], response['msg'])
def get_local_base(self, url=api._DISC_HTTP_URL):
response = self._session.get(f"{url}/api_version").json()
self.log.debug(f"<< Detected api {response['api_version']}")
return "%s%sv%s" % (url, response['api_base_url'], response['api_version'][0])
class Fbx():
Freebox object
def __init__(self, url=None, session=None):
self._trn = FbxTransport(url, session=session)
self.log = logging.getLogger("pyfbx.fbx")
# Create on the fly attributes to classes
_globals = globals()
for m_class in api.SYSTEMS:
setattr(self, m_class, _globals[m_class](self._trn))
for name, meth in api.SYSTEMS[m_class].items():
utils.add_class_func(getattr(self, m_class).__class__, name, meth)
def register(self, app_id, app_name, device):
Register app
self.app_id = app_id
data = {"app_id": self.app_id, "app_name": app_name, "device_name": device}
res = self._trn.api_exec("POST", "login/authorize/", data)
trackid, self.token = res["track_id"], res["app_token"]
s = "pending"
self.log.warning("Press Ok on the freebox to register application")
while s == "pending":
s = self._trn.api_exec("GET", f"login/authorize/{trackid}")["status"]
if s == "pending":
self.log.debug(f"Registration returned: {s}")
return s == "granted" and self.token
def mksession(self, app_id=None, token=None):
if token: # Don't overwrite previous token (used for refresh)
self.token = token
if app_id:
self.app_id = app_id
login = self._trn.api_exec("GET", "login/")
if not login['logged_in']:
data = {
"app_id": self.app_id,
"password":, "ascii"),
bytes(login['challenge'], "ascii"),
resp = self._trn.api_exec("POST", "login/session/", data)
session_token = resp["session_token"]
self.app_id = app_id
self.log.debug(f"Got session token, permissions are {resp['permissions']}")
return resp["permissions"]
class FbxErrorResponse(Exception):
def __init__(self, error_code, msg):
self.error_code = error_code
self.msg = msg
def __str__(self):
return f'{self.msg} [{self.error_code}]'
def logger():
log = logging.getLogger("pyfbx")
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
#stream_handler = logging.StreamHandler() only for main() and set by the user
#stream_handler.setFormatter(formatter) not in a library!
return log
log = logger()
# All FB subsystems are classes deriving from FbxClass
for _classname in api.SYSTEMS:
log.debug(f"Adding class {_classname} to locals")
locals()[_classname] = type(_classname, (FbxClass, ), {})