1""" 2Module to communicate with the AVM Fritz!Box. 3""" 4# This module is part of the FritzConnection package. 5# https://github.com/kbr/fritzconnection 6# License: MIT (https://opensource.org/licenses/MIT) 7# Author: Klaus Bremer 8 9 10import os 11import string 12import xml.etree.ElementTree as ElementTree 13 14import requests 15from requests.auth import HTTPDigestAuth 16 17from .devices import DeviceManager 18from .exceptions import ( 19 FritzConnectionException, 20 FritzServiceError, 21) 22from .soaper import Soaper 23 24# disable InsecureRequestWarning from urllib3 25# because of skipping certificate verification: 26import urllib3 27 28urllib3.disable_warnings() 29 30 31# FritzConnection defaults: 32FRITZ_IP_ADDRESS = "169.254.1.1" 33FRITZ_TCP_PORT = 49000 34FRITZ_TLS_PORT = 49443 35FRITZ_USERNAME = "dslf-config" # for Fritz!OS < 7.24 36FRITZ_IGD_DESC_FILE = "igddesc.xml" 37FRITZ_TR64_DESC_FILE = "tr64desc.xml" 38FRITZ_DESCRIPTIONS = [FRITZ_IGD_DESC_FILE, FRITZ_TR64_DESC_FILE] 39FRITZ_USERNAME_REQUIRED_VERSION = 7.24 40 41# same defaults as used by requests: 42DEFAULT_POOL_CONNECTIONS = 10 43DEFAULT_POOL_MAXSIZE = 10 44 45# supported protocols: 46PROTOCOLS = ['http://', 'https://'] 47 48 49class FritzConnection: 50 """ 51 Main class to set up a connection to the Fritz!Box router. All 52 parameters are optional. `address` should be the ip of a router, in 53 case that are multiple Fritz!Box routers in a network, the ip must 54 be given. Otherwise it is undefined which router will respond. If 55 `user` and `password` are not provided, the environment gets checked for 56 FRITZ_USERNAME and FRITZ_PASSWORD settings and taken from there, if 57 found. 58 59 The optional parameter `timeout` is a floating number in seconds 60 limiting the time waiting for a router response. This is a global 61 setting for the internal communication with the router. In case of a 62 timeout a `requests.ConnectTimeout` exception gets raised. 63 (`New in version 1.1`) 64 65 `use_tls` accepts a boolean for using encrypted communication with 66 the Fritz!Box. Default is `False`. 67 (`New in version 1.2`) 68 69 For some actions the Fritz!Box needs a password and since Fritz!OS 70 7.24 also requires a username, the previous default username is just 71 valid for OS versions < 7.24. In case the username is not given and 72 the system version is 7.24 or newer, FritzConnection uses the last 73 logged in username as default. 74 (`New in version 1.5`) 75 76 For applications where the urllib3 default connection-pool size 77 should get adapted, the arguments `pool_connections` and 78 `pool_maxsize` can get set explicitly. 79 (`New in version 1.6`) 80 """ 81 82 def __init__( 83 self, 84 address=None, 85 port=None, 86 user=None, 87 password=None, 88 timeout=None, 89 use_tls=False, 90 pool_connections=DEFAULT_POOL_CONNECTIONS, 91 pool_maxsize=DEFAULT_POOL_MAXSIZE, 92 ): 93 """ 94 Initialisation of FritzConnection: reads all data from the box 95 and also the api-description (the servicenames and according 96 actionnames as well as the parameter-types) that can vary among 97 models and stores these informations as instance-attributes. 98 This can be an expensive operation. Because of this an instance 99 of FritzConnection should be created once and reused in an 100 application. All parameters are optional. But if there is more 101 than one FritzBox in the network, an address (ip as string) must 102 be given, otherwise it is not defined which box may respond. If 103 no user is given the Environment gets checked for a 104 FRITZ_USERNAME setting. If there is no entry in the environment 105 the avm-default-username will be used. If no password is given 106 the Environment gets checked for a FRITZ_PASSWORD setting. So 107 password can be used without using configuration-files or even 108 hardcoding. The optional parameter `timeout` is a floating point 109 number in seconds limiting the time waiting for a router 110 response. The timeout can also be a tuple for different values 111 for connection- and read-timeout values: (connect timeout, read 112 timeout). The timeout is a global setting for the internal 113 communication with the router. In case of a timeout a 114 `requests.ConnectTimeout` exception gets raised. `use_tls` 115 accepts a boolean for using encrypted communication with the 116 Fritz!Box. Default is `False`. `pool_connections` and `pool_maxsize` 117 accept integers for changing the default urllib3 settings in order 118 to modify the number of reusable connections. 119 """ 120 if address is None: 121 address = FRITZ_IP_ADDRESS 122 if user is None: 123 user = os.getenv("FRITZ_USERNAME", FRITZ_USERNAME) 124 if password is None: 125 password = os.getenv("FRITZ_PASSWORD", "") 126 if port is None and use_tls: 127 port = FRITZ_TLS_PORT 128 elif port is None: 129 port = FRITZ_TCP_PORT 130 address = self.set_protocol(address, use_tls) 131 132 # a session will speed up connections (significantly for tls) 133 # and is required to change the default poolsize: 134 session = requests.Session() 135 session.verify = False 136 if password: 137 session.auth = HTTPDigestAuth(user, password) 138 adapter = requests.adapters.HTTPAdapter( 139 pool_connections=pool_connections, pool_maxsize=pool_maxsize) 140 session.mount(PROTOCOLS[use_tls], adapter) 141 # store as instance attributes for use by library modules 142 self.address = address 143 self.session = session 144 self.timeout = timeout 145 self.port = port 146 147 self.soaper = Soaper( 148 address, port, user, password, timeout=timeout, session=session 149 ) 150 self.device_manager = DeviceManager(timeout=timeout, session=session) 151 152 for description in FRITZ_DESCRIPTIONS: 153 source = f"{address}:{port}/{description}" 154 try: 155 self.device_manager.add_description(source) 156 except FritzConnectionException: 157 # resource not available: 158 # this can happen on devices not providing 159 # an igddesc-file. 160 # ignore this 161 pass 162 163 self.device_manager.scan() 164 self.device_manager.load_service_descriptions(address, port) 165 # set default user for FritzOS >= 7.24: 166 self._reset_user(user, password) 167 168 169 def __repr__(self): 170 """Return a readable representation""" 171 return ( 172 f"{self.modelname} at {self.soaper.address}\n" 173 f"FRITZ!OS: {self.system_version}" 174 ) 175 176 @property 177 def services(self): 178 """ 179 Dictionary of service instances. Keys are the service names. 180 """ 181 return self.device_manager.services 182 183 @property 184 def modelname(self): 185 """ 186 Returns the modelname of the router. 187 """ 188 return self.device_manager.modelname 189 190 @property 191 def system_version(self): 192 """ 193 Returns system version if known, otherwise None. 194 """ 195 return self.device_manager.system_version 196 197 @staticmethod 198 def normalize_name(name): 199 """ 200 Returns the normalized service name. E.g. WLANConfiguration and 201 WLANConfiguration:1 will get converted to WLANConfiguration1. 202 """ 203 if ":" in name: 204 name, number = name.split(":", 1) 205 name = name + number 206 elif name[-1] not in string.digits: 207 name = name + "1" 208 return name 209 210 @staticmethod 211 def set_protocol(url, use_tls): 212 """ 213 Sets the protocol of the `url` according to the `use_tls`-flag 214 and returns the modified `url`. Does not check whether the `url` 215 given as parameter is correct. 216 """ 217 url = url.split("//", 1)[-1] 218 return PROTOCOLS[use_tls] + url 219 220 def _reset_user(self, user, password): 221 """ 222 For Fritz!OS >= 7.24: if a password is given and the username is 223 the historic FRITZ_USERNAME, then check for the last logged-in 224 username and use this username for the soaper. Also recreate the 225 session used by the soaper and the device_manager. 226 227 This may not guarantee a valid user/password combination, but is 228 the way AVM recommends setting the required username in case a 229 username is not provided. 230 """ 231 try: 232 sys_version = float(self.system_version) 233 except (ValueError, TypeError): 234 # version not available: don't do anything 235 return 236 if (sys_version >= FRITZ_USERNAME_REQUIRED_VERSION 237 and user == FRITZ_USERNAME 238 and password 239 ): 240 last_user = None 241 response = self.call_action('LANConfigSecurity1', 'X_AVM-DE_GetUserList') 242 root = ElementTree.fromstring(response['NewX_AVM-DE_UserList']) 243 for node in root: 244 if node.tag == 'Username' and node.attrib['last_user'] == '1': 245 last_user = node.text 246 break 247 if last_user is not None: 248 self.session.auth = HTTPDigestAuth(last_user, password) 249 self.soaper.user = last_user 250 self.soaper.session = self.session 251 self.device_manager.session = self.session 252 253 254 # ------------------------------------------- 255 # public api: 256 # ------------------------------------------- 257 258 def call_action(self, service_name, action_name, *, arguments=None, **kwargs): 259 """ 260 Executes the given action of the given service. Both parameters 261 are required. Arguments are optional and can be provided as a 262 dictionary given to 'arguments' or as separate keyword 263 parameters. If 'arguments' is given additional 264 keyword-parameters as further arguments are ignored. 265 The argument values can be of type *str*, *int* or *bool*. 266 (Note: *bool* is provided since 1.3. In former versions 267 booleans must provided as numeric values: 1, 0). 268 If the service_name does not end with a number (like 1), a 1 269 gets added by default. If the service_name ends with a colon and a 270 number, the colon gets removed. So i.e. WLANConfiguration 271 expands to WLANConfiguration1 and WLANConfiguration:2 converts 272 to WLANConfiguration2. 273 Invalid service names will raise a ServiceError and invalid 274 action names will raise an ActionError. 275 """ 276 arguments = arguments if arguments else dict() 277 if not arguments: 278 arguments.update(kwargs) 279 service_name = self.normalize_name(service_name) 280 try: 281 service = self.device_manager.services[service_name] 282 except KeyError: 283 raise FritzServiceError(f'unknown service: "{service_name}"') 284 return self.soaper.execute(service, action_name, arguments) 285 286 def reconnect(self): 287 """ 288 Terminate the connection and reconnects with a new ip. 289 """ 290 self.call_action("WANIPConn1", "ForceTermination") 291 292 def reboot(self): 293 """ 294 Reboot the system. 295 """ 296 self.call_action("DeviceConfig1", "Reboot") 297