djib
/
pyfbx
mirror of https://framagit.org/djib/pyfbx
Watch 1
Star 0
Fork 0
Code Issues 0 Releases 9 Wiki Activity
Browse Source

Initial commit

tags/v0.0.2
Laurent Kislaire 1 year ago
commit
f662376b70
17 changed files with 871 additions and 0 deletions
Split View Show Diff Stats
  1. +112
    -0
      .gitignore
  2. +19
    -0
      LICENSE
  3. +2
    -0
      MANIFEST.in
  4. +68
    -0
      README.md
  5. +4
    -0
      pyfbx/__init__.py
  6. +254
    -0
      pyfbx/api.py
  7. +133
    -0
      pyfbx/client.py
  8. +47
    -0
      pyfbx/fb.pem
  9. +55
    -0
      pyfbx/mdns.py
  10. +49
    -0
      pyfbx/utils.py
  11. +5
    -0
      requirements-dev.txt
  12. +2
    -0
      requirements.txt
  13. +11
    -0
      setup.cfg
  14. +20
    -0
      setup.py
  15. +0
    -0
      tests/__init__.py
  16. +64
    -0
      tests/conn_test.py
  17. +26
    -0
      tests/sanity_test.py

+ 112
- 0
.gitignore View File

@@ -0,0 +1,112 @@
gen_api/
tests/token.txt
.buildlog
.dock
.pypt/gh-token
docs/.build/doctrees
docs/source
conf-priv/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class

# C extensions
*.so

# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST

# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec

# Installer logs
pip-log.txt
pip-delete-this-directory.txt

# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/

# Translations
*.mo
*.pot

# Django stuff:
*.log
local_settings.py
db.sqlite3

# Flask stuff:
instance/
.webassets-cache

# Scrapy stuff:
.scrapy

# Sphinx documentation
docs/_build/

# PyBuilder
target/

# Jupyter Notebook
.ipynb_checkpoints

# pyenv
.python-version

# celery beat schedule file
celerybeat-schedule

# SageMath parsed files
*.sage.py

# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/

# Spyder project settings
.spyderproject
.spyproject

# Rope project settings
.ropeproject

# mkdocs documentation
/site

# mypy
.mypy_cache/

+ 19
- 0
LICENSE View File

@@ -0,0 +1,19 @@
Copyright (c) 2019 Laurent Kislaire

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

+ 2
- 0
MANIFEST.in View File

@@ -0,0 +1,2 @@
graft pyfbx
global-exclude __pycache__ *.pyc

+ 68
- 0
README.md View File

@@ -0,0 +1,68 @@
# Freebox client library

## Features

* Full API coverage
* Https and Mdns discovery
* Thin client wrapper (~150 lines)

## Usage

The REST API is generated at runtime with the creation of class attributes for all Freebox subsystems.

```python
>>> from pyfbx import Fbx
>>> f=Fbx()
>>> f.<TAB>
f.Airmedia f.Download_Config f.Lan f.Rrd f.Upnpav
f.Call f.Download_Feeds f.Lcd f.Share f.Vpn
f.Connection f.Freeplug f.Nat f.Storage f.Vpn_Client
f.Contacts f.Fs f.Network_Share f.Switch f.Wifi
f.Dhcp f.Ftp f.Parental f.System f.mksession(
f.Download f.Igd f.Pvr f.Upload f.register(
>>> f.Freeplug.<TAB>
f.Freeplug.Get_a_particular_Freeplug_information(
f.Freeplug.Reset_a_Freeplug(
f.Freeplug.Get_the_current_Freeplugs_networks(

# Register application to get app token. Physical access is required.
>>> token = f.register("Id_SuperAppli", "Une superbe appli", "python")
# Generate session token
>>> f.mksession(app_id="Id_SupperAppli", token=token)

>>> help(f.Lan.Update_the_current_Lan_configuration)
Help on method Update_the_current_Lan_configuration:

Update_the_current_Lan_configuration(post_data) method of pyfbx.client.Lan instance
Update the current Lan configuration

Url parameters:
Post data:PostData

>>> help(f.Contacts.Access_a_given_contact_entry)
Help on method Access_a_given_contact_entry:

Access_a_given_contact_entry(id) method of pyfbx.client.Contacts instance
Access a given contact entry

Url parameters: id
Post data:

>>> f.Lan.Get_the_current_Lan_configuration()
{'name_dns': 'freebox-server', 'name': 'Freebox Server', 'name_mdns': 'Freebox-Server',
'mode': 'router', 'name_netbios': 'Freebox_Server', 'ip': '192.168.1.254'}

# Any subsequent call to mksession will refresh the session token
>>> f.mksession()
```

## Testing

You can run tests with

```shell
pip3 install -r requirements-dev.txt
pytest tests
```
Currently to get 100% coverage, a Freebox sitting on its default IP (192.168.1.254) is required.


+ 4
- 0
pyfbx/__init__.py View File

@@ -0,0 +1,4 @@
"""
Freebox client library
"""
from .client import *

+ 254
- 0
pyfbx/api.py View File

@@ -0,0 +1,254 @@
"""
Module dedicated to API spec
"""
from collections import namedtuple
from . import utils


_DISC_HTTP_URL = "http://mafreebox.freebox.fr"
_DISC_MDNS_NAME = "_fbx-api._tcp.local."


class _M(namedtuple("Method", ('url', 'doc', 'args'), defaults=([],))):
"""
Accessor class for API methods
"""
__slots__ = ()

@property
def http_method(self):
return self.url.split()[0]

@property
def endpoint(self):
return self.url.split()[1]

def __str__(self):
return "%s\n\nUrl parameters: %s\nPost data:%s" % (
self.doc, ",".join(utils.get_params(self.endpoint)), ",".join(self.args))


SYSTEMS = {
"Download": {
"Retrieve_a_Download_task": _M("GET downloads/", "Retrieve a Download task"),
"Delete_a_Download_task": _M("DELETE downloads/{id}", "Delete a Download task"),
"Update_a_Download_task": _M("PUT downloads/{id}", "Update a Download task", ["PostData"]),
"Get_download_log": _M("GET downloads/{id}/log", "Get download log"),
"Adding_a_new_Download_task": _M("POST downloads/add", "Adding a new Download task", ["PostData"]),
},
"Download_Feeds": {
"Get_the_list_of_all_download_Feeds": _M("GET downloads/feeds/", "Get the list of all download Feeds"),
"Get_a_download_Feed": _M("GET downloads/feeds/{id}", "Get a download Feed"),
"Add_a_Download_Feed": _M("POST downloads/feeds/", "Add a Download Feed", ["PostData"]),
"Delete_Download_Feed": _M("DELETE downloads/feeds/{id}", "Delete Download Feed"),
"Update_a_Download_Feed": _M("PUT downloads/feeds/{id}", "Update a Download Feed", ["PostData"]),
"Refresh_a_Download_Feed": _M("POST downloads/feeds/{id}/fetch", "Refresh a Download Feed", ["PostData"]),
"Refresh_all_Download_Feeds": _M("POST downloads/feeds/fetch", "Refresh all Download Feeds", ["PostData"]),
"Get_the_items_of_a_given_RSS_feed": _M("GET downloads/feeds/{feed_id}/items/", "Get the items of a given RSS feed"),
"Update_a_feed_item": _M("PUT downloads/feeds/{feed_id}/items/{item_id}", "Update a feed item", ["PostData"]),
"Download_a_feed_item": _M("POST downloads/feeds/{feed_id}/items/{item_id}/download", "Download a feed item", ["PostData"]),
"Mark_all_items_as_read": _M("POST downloads/feeds/{feed_id}/items/mark_all_as_read", "Mark all items as read", ["PostData"]),
},
"Download_Config": {
"Updating_the_download_config": _M("PUT downloads/config/", "Updating the Download Config", ["PostData"]),
"Updating_the_current_Throttling_mode": _M("PUT downloads/throttling", "Updating the current Throttling mode", ["PostData"]),
},
"Fs": {
"List_every_tasks": _M("GET fs/tasks/", "List every tasks"),
"List_a_task": _M("GET fs/tasks/{id}", "List a task"),
"Delete_a_task": _M("DELETE fs/tasks/{id}", "Delete a task"),
"Update_a_task": _M("PUT fs/tasks/{id}", "Update a task", ["PostData"]),
"List_files": _M("GET fs/ls/{path}", "List files"),
"Get_file_information": _M("GET fs/info/{path}", "Get file information"),
"Conflict_resolution": _M("POST fs/mv/", "Conflict resolution", ["PostData"]),
"Move_files": _M("POST fs/mv/", "Move files", ["PostData"]),
"Copy_files": _M("POST fs/cp/", "Copy files", ["PostData"]),
"Remove_files": _M("POST fs/rm/", "Remove files", ["PostData"]),
"Cat_files": _M("POST fs/cat/", "Cat files", ["PostData"]),
"Create_an_archive": _M("POST fs/archive/", "Create an archive", ["PostData"]),
"Extract_a_file": _M("POST fs/extract/", "Extract a file", ["PostData"]),
"Repair_a_file": _M("POST fs/repair/", "Repair a file", ["PostData"]),
"Hash_a_file": _M("POST fs/hash/", "Hash a file", ["PostData"]),
"Create_a_directory": _M("POST fs/mkdir/", "Create a directory", ["PostData"]),
"Rename_a_file/folder": _M("POST fs/rename/", "Rename a file/folder", ["PostData"]),
"Download_a_file": _M("GET dl/{path}", "Download a file"),
},
"Share": {
"Retrieve_a_File_Sharing_link": _M("GET share_link/", "Retrieve a File Sharing link"),
"Delete_a_File_Sharing_link": _M("DELETE share_link/{token}", "Delete a File Sharing link"),
"Create_a_File_Sharing_link": _M("POST share_link/", "Create a File Sharing link", ["PostData"]),
},
"Upload": {
"File_Upload_example": _M("GET ws/upload", "File Upload example"),
"Get_the_list_of_uploads": _M("GET upload/", "Get the list of uploads"),
"Track_an_upload_status": _M("GET upload/{id}", "Track an upload status"),
"Cancel_an_upload": _M("DELETE upload/{id}/cancel", "Cancel an upload"),
"Delete_an_upload": _M("DELETE upload/{id}", "Delete an upload"),
"Cleanup_all_terminated_uploads": _M("DELETE upload/clean", "Cleanup all terminated uploads"),
},
"Airmedia": {
"Get_the_current_AirMedia_configuration": _M("GET airmedia/config/", "Get the current AirMedia configuration"),
"Update_the_current_AirMedia_configuration": _M("PUT airmedia/config/", "Update the current AirMedia configuration", ["PostData"]),
"Get_the_list_of_available_AirMedia_receivers": _M("GET airmedia/receivers/", "Get the list of available AirMedia receivers"),
},
"Rrd": {
},
"Call": {
"List_every_calls": _M("GET call/log/", "List every calls"),
"Delete_every_calls": _M("POST call/log/delete_all/", "Delete every calls", ["PostData"]),
"Mark_every_calls_as_read": _M("POST call/log/mark_all_as_read/", "Mark every calls as read", ["PostData"]),
"Access_a_given_call_entry": _M("GET call/log/{id}", "Access a given call entry"),
"Delete_a_call": _M("DELETE call/log/{id}", "Delete a call"),
"Update_a_call_entry": _M("PUT call/log/{id}", "Update a call entry", ["PostData"]),
},
"Contacts": {
"Get_a_list_of_contacts": _M("GET contact/", "Get a list of contacts"),
"Access_a_given_contact_entry": _M("GET contact/{id}", "Access a given contact entry"),
"Create_a_contact": _M("POST contact/", "Create a contact", ["PostData"]),
"Delete_a_contact": _M("DELETE contact/{id}", "Delete a contact"),
"Update_a_contact_entry": _M("PUT contact/{id}", "Update a contact entry", ["PostData"]),
"Get_the_list_of_numbers_for_a_given_contact": _M("GET contact/{contact_id}/[numbers|addresses|urls|emails]/", "Get the list of numbers for a given contact"),
"Access_a_given_contact_number": _M("GET [number,address,url,email]/{id}", "Access a given contact number"),
"Create_a_contact_number": _M("POST [number,address,url,email]/", "Create a contact number", ["PostData"]),
"Delete_a_contact_number": _M("DELETE [number,address,url,email]/{id}", "Delete a contact number"),
"Update_a_contact_number": _M("PUT [number,address,url,email]/{id}", "Update a contact number", ["PostData"]),
},
"Connection": {
"Get_the_current_Connection_status": _M("GET connection/", "Get the current Connection status"),
"Get_the_current_Connection_configuration": _M("GET connection/config/", "Get the current Connection configuration"),
"Update_the_Connection_configuration": _M("PUT connection/config/", "Update the Connection configuration", ["PostData"]),
"Get_the_current_IPv6_Connection_configuration": _M("GET connection/ipv6/config/", "Get the current IPv6 Connection configuration"),
"Update_the_IPv6_Connection_configuration": _M("PUT connection/ipv6/config/", "Update the IPv6 Connection configuration", ["PostData"]),
"Get_the_current_xDSL_infos": _M("GET connection/xdsl/", "Get the current xDSL infos"),
"Get_the_current_FTTH_status": _M("GET connection/ftth/", "Get the current FTTH status"),
"Get_the_status_of_a_DynDNS_service": _M("GET connection/ddns/{provider}/status/", "Get the status of a DynDNS service"),
"Get_the_config_of_a_DynDNS_service": _M("GET connection/ddns/{provider}/", "Get the config of a DynDNS service"),
"Set_the_config_of_a_DynDNS_service": _M("PUT connection/ddns/{provider}/", "Set the config of a DynDNS service", ["PostData"]),
},
"Lan": {
"Get_the_current_Lan_configuration": _M("GET lan/config/", "Get the current Lan configuration"),
"Update_the_current_Lan_configuration": _M("PUT lan/config/", "Update the current Lan configuration", ["PostData"]),
},
"Freeplug": {
"Get_the_current_Freeplugs_networks": _M("GET freeplug/", "Get the current Freeplugs networks"),
"Get_a_particular_Freeplug_information": _M("GET freeplug/{id}/", "Get a particular Freeplug information"),
"Reset_a_Freeplug": _M("POST freeplug/{id}/reset/", "Reset a Freeplug"),
},
"Dhcp": {
"Get_the_current_DHCP_configuration": _M("GET dhcp/config/", "Get the current DHCP configuration"),
"Update_the_current_DHCP_configuration": _M("PUT dhcp/config/", "Update the current DHCP configuration", ["PostData"]),
"Get_the_list_of_DHCP_static_leases": _M("GET dhcp/static_lease/", "Get the list of DHCP static leases"),
"Get_a_given_DHCP_static_lease": _M("GET dhcp/static_lease/{id}", "Get a given DHCP static lease"),
"Update_DHCP_static_lease": _M("PUT dhcp/static_lease/{id}", "Update DHCP static lease", ["PostData"]),
"Delete_a_DHCP_static_lease": _M("DELETE dhcp/static_lease/{id}", "Delete a DHCP static lease"),
"Add_a_DHCP_static_lease": _M("POST dhcp/static_lease/", "Add a DHCP static lease", ["PostData"]),
"Get_the_list_of_DHCP_dynamic_leases": _M("GET dhcp/dynamic_lease/", "Get the list of DHCP dynamic leases"),
},
"Ftp": {
"Get_the_current_Ftp_configuration": _M("GET ftp/config/", "Get the current Ftp configuration"),
"Update_the_FTP_configuration": _M("PUT ftp/config/", "Update the FTP configuration", ["PostData"]),
},
"Nat": {
"Get_the_current_Dmz_configuration": _M("GET fw/dmz/", "Get the current Dmz configuration"),
"Update_the_current_Dmz_configuration": _M("PUT fw/dmz/", "Update the current Dmz configuration", ["PostData"]),
},
"Igd": {
"Get_the_current_UPnP_IGD_configuration": _M("GET upnpigd/config/", "Get the current UPnP IGD configuration"),
"Update_the_UPnP_IGD_configuration": _M("PUT upnpigd/config/", "Update the UPnP IGD configuration", ["PostData"]),
"Get_the_list_of_current_redirection": _M("GET upnpigd/redir/", "Get the list of current redirection"),
"Delete_a_redirection": _M("DELETE upnpigd/redir/{id}", "Delete a redirection"),
},
"Lcd": {
"Get_the_current_LCD_configuration": _M("GET lcd/config/", "Get the current LCD configuration"),
"Update_the_lcd_configuration": _M("PUT lcd/config/", "Update the lcd configuration", ["PostData"]),
},
"Network_Share": {
"Get_the_current_Samba_configuration": _M("GET netshare/samba/", "Get the current Samba configuration"),
"Update_the_Samba_configuration": _M("PUT netshare/samba/", "Update the Samba configuration", ["PostData"]),
"Get_the_current_Afp_configuration": _M("GET netshare/afp/", "Get the current Afp configuration"),
"Update_the_Afp_configuration": _M("PUT netshare/afp/", "Update the Afp configuration", ["PostData"]),
},
"Upnpav": {
"Get_the_current_UPnP_AV_configuration": _M("GET upnpav/config/", "Get the current UPnP AV configuration"),
"Update_the_UPnP_AV_configuration": _M("PUT upnpav/config/", "Update the UPnP AV configuration", ["PostData"]),
},
"Switch": {
"Get_the_current_switch_status": _M("GET switch/status/", "Get the current switch status"),
"Get_a_port_configuration": _M("GET switch/port/{id}", "Get a port configuration"),
"Update_a_port_configuration": _M("PUT switch/port/{id}", "Update a port configuration", ["PostData"]),
"Get_a_port_stats": _M("GET switch/port/{id}/stats", "Get a port stats"),
},
"Wifi": {
"Get_the_current_Wi-Fi_global_configuration": _M("GET wifi/config/", "Get the current Wi-Fi global configuration"),
"Update_the_Wi-Fi_global_configuration": _M("PUT wifi/config/", "Update the Wi-Fi global configuration", ["PostData"]),
"Wi-Fi_AP_API": _M("GET wifi/ap/", "Wi-Fi AP API"),
"Get_Wi-Fi_Stations_List": _M("GET wifi/ap/{id}/stations/", "Get Wi-Fi Stations List"),
"Wi-Fi_BSS_API": _M("GET wifi/bss/", "Wi-Fi BSS API"),
"List_AP_neighbors": _M("GET wifi/ap/{id}/neighbors/", "List AP neighbors"),
"List_Wi-Fi_channels_usage": _M("GET wifi/ap/{id}/channel_usage/", "List Wi-Fi channels usage"),
"Refresh_radar_informations": _M("POST wifi/ap/{id}/neighbors/scan", "Refresh radar informations", ["PostData"]),
"Get_Wi-Fi_Planning": _M("GET wifi/planning/", "Get Wi-Fi Planning"),
"Update_Wi-Fi_Planning": _M("PUT wifi/planning/", "Update Wi-Fi Planning", ["PostData"]),
"Get_the_MAC_filter_list": _M("GET wifi/mac_filter/", "Get the MAC filter list"),
"Getting_a_particular_MAC_filter": _M("GET wifi/mac_filter/{filter_id}", "Getting a particular MAC filter"),
"Updating_a_MAC_filter": _M("PUT wifi/mac_filter/{filter_id}", "Updating a MAC filter", ["PostData"]),
"Delete_a_MAC_filter": _M("DELETE wifi/mac_filter/{filter_id}", "Delete a MAC filter"),
"Create_a_new_MAC_filter": _M("POST wifi/mac_filter/", "Create a new MAC filter", ["PostData"]),
},
"System": {
"Get_the_current_system_info": _M("GET system/", "Get the current system info [UNSTABLE]"),
"Reboot_the_system": _M("POST system/reboot/", "Reboot the system", ["PostData"]),
},
"Vpn": {
"VPN_Server_List_API": _M("GET vpn/", "VPN Server List API"),
"Get_a_VPN_config": _M("GET vpn/{vpn_id}/config/", "Get a VPN config"),
"Update_the_VPN_configuration": _M("PUT vpn/openvpn_routed/config/", "Update the VPN configuration", ["PostData"]),
"VPN_Server_User_List": _M("GET vpn/user/", "VPN Server User List"),
"Get_a_VPN_user": _M("GET vpn/user/{login}", "Get a VPN user"),
"Add_a_VPN_User": _M("POST vpn/user/", "Add a VPN User", ["PostData"]),
"Delete_a_VPN_User": _M("DELETE vpn/user/{login}", "Delete a VPN User"),
"Update_a_VPN_User": _M("PUT vpn/user/{login}", "Update a VPN User", ["PostData"]),
"Get_the_VPN_server_IP_pool_reservations": _M("GET vpn/ip_pool/", "Get the VPN server IP pool reservations"),
"Get_the_list_of_connections": _M("GET vpn/connection/", "Get the list of connections"),
"Close_a_given_connection": _M("DELETE vpn/connection/{id}", "Close a given connection"),
"Donwload_a_user_configuration_file": _M("GET vpn/download_config/{server_name}/{login}", "Donwload a user configuration file"),
},
"Vpn_Client": {
"Get_VPN_Client_configuration_list": _M("GET vpn_client/config/", "Get VPN Client configuration list"),
"Get_a_VPN_client_config": _M("GET vpn_client/config/{id}", "Get a VPN client config"),
"Add_a_VPN_client_configuration": _M("POST vpn_client/config/", "Add a VPN client configuration", ["PostData"]),
"Delete_a_VPN_client_Configuration": _M("DELETE vpn_client/config/{id}", "Delete a VPN client Configuration"),
"Update_the_VPN_client_configuration": _M("PUT vpn_client/config/{id}", "Update the VPN client configuration", ["PostData"]),
"Get_the_VPN_client_status": _M("GET vpn_client/status", "Get the VPN client status"),
"Get_the_VPN_client_logs": _M("GET vpn_client/log", "Get the VPN client logs"),
},
"Storage": {
"Get_the_list_of_disks": _M("GET storage/disk/", "Get the list of disks"),
"Get_a_given_disk_info": _M("GET storage/disk/{id}", "Get a given disk info"),
"Update_a_disk_state": _M("PUT storage/disk/{id}", "Update a disk state", ["PostData"]),
"Format_a_disk": _M("PUT storage/disk/{id}/format/", "Format a disk", ["PostData"]),
"Get_the_list_of_partitions": _M("GET storage/partition/", "Get the list of partitions"),
"Get_a_given_partition_info": _M("GET storage/partition/{id}", "Get a given partition info"),
"Update_a_partition_state": _M("PUT storage/partition/{id}", "Update a partition state", ["PostData"]),
"Check_a_partition": _M("PUT storage/partition/{id}/check/", "Check a partition", ["PostData"]),
"Get_the_current_storage_configuration": _M("GET storage/config/", "Get the current storage configuration"),
"Update_the_External_Storage_configuration": _M("PUT storage/config/", "Update the External Storage configuration", ["PostData"]),
},
"Parental": {
"Get_parental_filter_config": _M("GET parental/config/", "Get parental filter config"),
"Update_parental_filter_config": _M("PUT parental/config/", "Update parental filter config", ["PostData"]),
"Retrieve_all_Parental_Filter_rules": _M("GET parental/filter/", "Retrieve all Parental Filter rules"),
"Retrieve_a_Parental_Filter_rule": _M("GET parental/filter/{id}", "Retrieve a Parental Filter rule"),
"Delete_a_Parental_Filter_rule": _M("DELETE parental/filter/{id}", "Delete a Parental Filter rule"),
"Update_a_Parental_Filter_rule": _M("PUT parental/filter/{id}", "Update a Parental Filter rule", ["PostData"]),
"Add_a_Parental_Filter_rule": _M("POST parental/filter/", "Add a Parental Filter rule", ["PostData"]),
"Get_a_parental_filter_Planning": _M("GET parental/filter/{id}/planning", "Get a parental filter Planning"),
"Update_a_parental_filter_Planning": _M("PUT parental/filter/{id}/planning", "Update a parental filter Planning", ["PostData"]),
},
"Pvr": {
"Get_the_current_PVR_configuration": _M("GET pvr/config/", "Get the current PVR configuration"),
"Update_the_current_PVR_configuration": _M("PUT pvr/config/", "Update the current PVR configuration", ["PostData"]),
"Getting_the_current_quota_info": _M("GET pvr/quota/", "Getting the current quota info"),
"Request_next_quota_threshold": _M("PUT pvr/quota/", "Request next quota threshold", ["PostData"]),
},
}

+ 133
- 0
pyfbx/client.py View File

@@ -0,0 +1,133 @@
#!/usr/bin/python3
"""
Fbx Client
"""
import os
import re
import hmac
import hashlib
import time
import requests
from . import api
from . import utils
from . import mdns

__all__ = [
"FbxTransport",
"Fbx",
"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.set_url(url)

def set_url(self, url):
if url is None:
# Detect using MDNS or fallback
self._url = mdns.FbxMDNS().search() or self.get_local_base()
else:
if re.search("https?://", url) is None:
url = "http://" + url
if "/api/" not in url:
self._url = self.get_local_base(url)
else:
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)
req_response.raise_for_status()
response = req_response.json()
if response['success']:
if 'result' in response:
return response['result']
else:
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()
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)

# 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"
while s == "pending":
s = self._trn.api_exec("GET", f"login/authorize/{trackid}")["status"]
if s == "pending":
time.sleep(1)
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": hmac.new(bytes(self.token, "ascii"),
bytes(login['challenge'], "ascii"),
hashlib.sha1).hexdigest()
}
resp = self._trn.api_exec("POST", "login/session/", data)
session_token = resp["session_token"]
self.app_id = app_id
self._trn.set_session_header(session_token)
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}]'


# All FB subsystems are classes deriving from FbxClass
for _classname in api.SYSTEMS:
locals()[_classname] = type(_classname, (FbxClass, ), {})
__all__.append(_classname)

+ 47
- 0
pyfbx/fb.pem View File

@@ -0,0 +1,47 @@
-----BEGIN CERTIFICATE-----
MIICWTCCAd+gAwIBAgIJAMaRcLnIgyukMAoGCCqGSM49BAMCMGExCzAJBgNVBAYT
AkZSMQ8wDQYDVQQIDAZGcmFuY2UxDjAMBgNVBAcMBVBhcmlzMRMwEQYDVQQKDApG
cmVlYm94IFNBMRwwGgYDVQQDDBNGcmVlYm94IEVDQyBSb290IENBMB4XDTE1MDkw
MTE4MDIwN1oXDTM1MDgyNzE4MDIwN1owYTELMAkGA1UEBhMCRlIxDzANBgNVBAgM
BkZyYW5jZTEOMAwGA1UEBwwFUGFyaXMxEzARBgNVBAoMCkZyZWVib3ggU0ExHDAa
BgNVBAMME0ZyZWVib3ggRUNDIFJvb3QgQ0EwdjAQBgcqhkjOPQIBBgUrgQQAIgNi
AASCjD6ZKn5ko6cU5Vxh8GA1KqRi6p2GQzndxHtuUmwY8RvBbhZ0GIL7bQ4f08ae
JOv0ycWjEW0fyOnAw6AYdsN6y1eNvH2DVfoXQyGoCSvXQNAUxla+sJuLGICRYiZz
mnijYzBhMB0GA1UdDgQWBBTIB3c2GlbV6EIh2ErEMJvFxMz/QTAfBgNVHSMEGDAW
gBTIB3c2GlbV6EIh2ErEMJvFxMz/QTAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB
/wQEAwIBhjAKBggqhkjOPQQDAgNoADBlAjA8tzEMRVX8vrFuOGDhvZr7OSJjbBr8
gl2I70LeVNGEXZsAThUkqj5Rg9bV8xw3aSMCMQCDjB5CgsLH8EdZmiksdBRRKM2r
vxo6c0dSSNrr7dDN+m2/dRvgoIpGL2GauOGqDFY=
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
MIIFmjCCA4KgAwIBAgIJAKLyz15lYOrYMA0GCSqGSIb3DQEBCwUAMFoxCzAJBgNV
BAYTAkZSMQ8wDQYDVQQIDAZGcmFuY2UxDjAMBgNVBAcMBVBhcmlzMRAwDgYDVQQK
DAdGcmVlYm94MRgwFgYDVQQDDA9GcmVlYm94IFJvb3QgQ0EwHhcNMTUwNzMwMTUw
OTIwWhcNMzUwNzI1MTUwOTIwWjBaMQswCQYDVQQGEwJGUjEPMA0GA1UECAwGRnJh
bmNlMQ4wDAYDVQQHDAVQYXJpczEQMA4GA1UECgwHRnJlZWJveDEYMBYGA1UEAwwP
RnJlZWJveCBSb290IENBMIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEA
xqYIvq8538SH6BJ99jDlOPoyDBrlwKEp879oYplicTC2/p0X66R/ft0en1uSQadC
sL/JTyfgyJAgI1Dq2Y5EYVT/7G6GBtVH6Bxa713mM+I/v0JlTGFalgMqamMuIRDQ
tdyvqEIs8DcfGB/1l2A8UhKOFbHQsMcigxOe9ZodMhtVNn0mUyG+9Zgu1e/YMhsS
iG4Kqap6TGtk80yruS1mMWVSgLOq9F5BGD4rlNlWLo0C3R10mFCpqvsFU+g4kYoA
dTxaIpi1pgng3CGLE0FXgwstJz8RBaZObYEslEYKDzmer5zrU1pVHiwkjsgwbnuy
WtM1Xry3Jxc7N/i1rxFmN/4l/Tcb1F7x4yVZmrzbQVptKSmyTEvPvpzqzdxVWuYi
qIFSe/njl8dX9v5hjbMo4CeLuXIRE4nSq2A7GBm4j9Zb6/l2WIBpnCKtwUVlroKw
NBgB6zHg5WI9nWGuy3ozpP4zyxqXhaTgrQcDDIG/SQS1GOXKGdkCcSa+VkJ0jTf5
od7PxBn9/TuN0yYdgQK3YDjD9F9+CLp8QZK1bnPdVGywPfL1iztngF9J6JohTyL/
VMvpWfS/X6R4Y3p8/eSio4BNuPvm9r0xp6IMpW92V8SYL0N6TQQxzZYgkLV7TbQI
Hw6v64yMbbF0YS9VjS0sFpZcFERVQiodRu7nYNC1jy8CAwEAAaNjMGEwHQYDVR0O
BBYEFD2erMkECujilR0BuER09FdsYIebMB8GA1UdIwQYMBaAFD2erMkECujilR0B
uER09FdsYIebMA8GA1UdEwEB/wQFMAMBAf8wDgYDVR0PAQH/BAQDAgGGMA0GCSqG
SIb3DQEBCwUAA4ICAQAZ2Nx8mWIWckNY8X2t/ymmCbcKxGw8Hn3BfTDcUWQ7GLRf
MGzTqxGSLBQ5tENaclbtTpNrqPv2k6LY0VjfrKoTSS8JfXkm6+FUtyXpsGK8MrLL
hZ/YdADTfbbWOjjD0VaPUoglvo2N4n7rOuRxVYIij11fL/wl3OUZ7GHLgL3qXSz0
+RGW+1oZo8HQ7pb6RwLfv42Gf+2gyNBckM7VVh9R19UkLCsHFqhFBbUmqwJgNA2/
3twgV6Y26qlyHXXODUfV3arLCwFoNB+IIrde1E/JoOry9oKvF8DZTo/Qm6o2KsdZ
dxs/YcIUsCvKX8WCKtH6la/kFCUcXIb8f1u+Y4pjj3PBmKI/1+Rs9GqB0kt1otyx
Q6bqxqBSgsrkuhCfRxwjbfBgmXjIZ/a4muY5uMI0gbl9zbMFEJHDojhH6TUB5qd0
JJlI61gldaT5Ci1aLbvVcJtdeGhElf7pOE9JrXINpP3NOJJaUSueAvxyj/WWoo0v
4KO7njox8F6jCHALNDLdTsX0FTGmUZ/s/QfJry3VNwyjCyWDy1ra4KWoqt6U7SzM
d5jENIZChM8TnDXJzqc+mu00cI3icn9bV9flYCXLTIsprB21wVSMh0XeBGylKxeB
S27oDfFq04XSox7JM9HdTt2hLK96x1T7FpFrBTnALzb7vHv9MhXqAT90fPR/8A==
-----END CERTIFICATE-----

+ 55
- 0
pyfbx/mdns.py View File

@@ -0,0 +1,55 @@
"""
Search Freebox through MDNS
"""

import time
from zeroconf import ServiceBrowser, Zeroconf
from . import api


class FbxMDNS():
"""
Search for a freebox using MDNS
"""
class MyListener:
"""
Callback class to get information of the available MDNS service
"""

def __init__(self):
self._info = None

def add_service(self, zeroconf, stype, name):
info = zeroconf.get_service_info(stype, name)
self._info = info

@property
def svc_info(self):
return self._info

def __init__(self, timeout=1):
self.timeout = timeout

def search(self, svc_name=api._DISC_MDNS_NAME, timeout=1):
zeroconf = Zeroconf()
self._listener = FbxMDNS.MyListener()
browser = ServiceBrowser(zeroconf, svc_name, self._listener)
time.sleep(timeout)
browser.cancel() # slow
zeroconf.close() # slow
if self.svc_info:
prop = self.svc_prop
base = "%sv%s" % (prop['api_base_url'], prop['api_version'][0])
if int(prop['https_available']):
return f"https://{prop['api_domain']}:{prop['https_port']}{base}"
else:
return f"http://{prop['api_domain']}{base}"

@property
def svc_info(self):
return self._listener.svc_info

@property
def svc_prop(self):
return {k.decode(): v.decode()
for k, v in self._listener.svc_info.properties.items()}

+ 49
- 0
pyfbx/utils.py View File

@@ -0,0 +1,49 @@
"""
Common utils
"""


class Default(dict):
"""
Subclass dict to track __missing__ elements
"""

def __init__(self):
dict.__init__(self)
self._missing_keys = []

def __missing__(self, key):
self._missing_keys.append(key)
return key

@property
def missing_keys(self):
return self._missing_keys


def get_params(url):
"""
Extract parameters from f-strings
"""
f_params = Default()
url.format_map(f_params)
return f_params.missing_keys


def add_class_func(cls, name, method):
"""
Attach to class a callback to transport with corresponding API endpoint and parameters
"""

params = (["post_data"] if method.args else []) + get_params(method.endpoint)
input_params = ", ".join(params)
passed_param = ", ".join([f"{e}={e}" for e in params])
namespace = {}
code = (f"def func(self, {input_params}):\n"
f" return self._trn.api_exec(\n"
f" \"{method.http_method}\", \"{method.endpoint}\", {passed_param})")
exec(code, namespace)
func = namespace['func']
func.__name__ = name
func.__doc__ = str(method)
setattr(cls, name, func)

+ 5
- 0
requirements-dev.txt View File

@@ -0,0 +1,5 @@
mock
pytest
pytest-cov
twine
wheel

+ 2
- 0
requirements.txt View File

@@ -0,0 +1,2 @@
requests
zeroconf

+ 11
- 0
setup.cfg View File

@@ -0,0 +1,11 @@
[tool:pytest]
norecursedirs = .git
addopts = --cov --cov-report term-missing

[coverage:run]
branch = True
source = pyfbx
omit = tests/*

[wheel]
universal = 0

+ 20
- 0
setup.py View File

@@ -0,0 +1,20 @@
import pathlib
from setuptools import setup, find_packages

setup(
name="pyfbx",
version="0.0.2",
description="Freebox thin client",
long_description=(pathlib.Path(__file__).parent / "README.md").read_text(),
long_description_content_type='text/markdown',
author_email="teebeenator@gmail.com",
url='https://framagit.org/sun/pyfbx',
license="MIT",
classifiers=[
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
],
packages=find_packages(),
install_requires=["zeroconf", "requests"],
include_package_data=True,
)

+ 0
- 0
tests/__init__.py View File


+ 64
- 0
tests/conn_test.py View File

@@ -0,0 +1,64 @@
#!/usr/bin/env python3

import time
import os
import sys
import requests
import json
import pytest
from mock import (patch, PropertyMock, MagicMock)
from pyfbx import Fbx
from pyfbx.client import (FbxErrorResponse, FbxTransport)
from pyfbx.mdns import FbxMDNS


def test_fbx_register():
f = Fbx("http://192.168.1.254/api/v6")
with patch('pyfbx.client.FbxTransport.api_exec',
side_effect=iter([
{"track_id": "id", "app_token": "tok"},
{"status": "pending"},
{"status": "granted"}])):
f.register(app_id="id", app_name="name", device="device")


@pytest.mark.local_fb
def test_fbx_session_local():
tokname = os.path.join(os.path.dirname(__file__), 'token.txt')

with open(tokname) as tokfile:
token = tokfile.readline().rstrip()
app_id = tokfile.readline().rstrip()
print(f"token:{token}, app_id:{app_id}")
f = Fbx()
f.mksession(app_id=app_id, token=token)
r = f.Contacts.Create_a_contact(post_data={"display_name": "Sandy Kilo",
"first_name": "Sandy",
"last_name": "Kilo"})
f.Contacts.Delete_a_contact(id=r['id'])
with pytest.raises(FbxErrorResponse):
try:
f.Contacts.Access_a_given_contact_entry(1)
except FbxErrorResponse as e:
print(e)
raise
with patch('requests.Response', PropertyMock) as mock_resp:
mock_resp.raise_for_status = MagicMock()
mock_resp.json = MagicMock()
f.Contacts.Get_a_list_of_contacts()
f.mksession()


def test_mdns():
prop = {
'https_available': '0',
'api_base_url': '/api',
'api_version': '6.6',
'api_domain': 'example.com'}

m = FbxMDNS()
m.search(svc_name="_foobar._tcp.local.")
with patch('pyfbx.mdns.FbxMDNS.svc_prop', new_callable=PropertyMock) as mock_prop:
mock_prop.return_value = prop
m.search(svc_name="_foobar._tcp.local.")
m.search()

+ 26
- 0
tests/sanity_test.py View File

@@ -0,0 +1,26 @@
#!/usr/bin/env python3

import time
import sys
import requests
import pytest
from mock import patch, MagicMock

from pyfbx import Fbx


@pytest.mark.local_fb
def test_fbx_hardcoded_url():
f = Fbx()
assert isinstance(f, Fbx)
with patch('pyfbx.mdns.FbxMDNS.search', return_value=None):
f = Fbx()
f = Fbx(url="http://12.34.56.78/api/v4")
f = Fbx(url="12.34.56.78/api/v4")
f = Fbx(url="http://192.168.1.254")


@pytest.mark.local_fb
def test_fbx_mdns():
f = Fbx()
f = Fbx(session=requests.Session())

Write Preview
Loading…
Cancel
Save