TR069 Honeyclient API Reference

TR-069 Client

class tr069.Client(acs_url: str, device: tr069.data.device.Device = Device(AVM FRITZ!Box, Serial: 0896D776FAA2, 11 parameters), *, log: bool = True, basic_auth: typing.Union[typing.Tuple[str, str], NoneType] = None, digest_auth: typing.Union[typing.Tuple[str, str], NoneType] = None, cert: typing.Union[typing.Tuple[str, str], str] = None, **requests_kwargs)

A TR-069 Client instance to interact with an ACS.

Parameters:
  • acs_url – The ACS URL.
  • device – The device represented by the client.
  • log – If True, all requests and responses are logged to stdout.
  • basic_auth – A (user, pass) tuple used for HTTP basic authentication.
  • digest_auth – A (user, pass) tuple used for HTTP digest authentication.
  • cert – TLS Client Certificate, see http://docs.python-requests.org/en/master/user/advanced/#ssl-cert-verification
  • **requests_kwargs – Additional arguments passed to subsequent internal calls of requests.post().
request(data: str, *, fix_cwmp_id: bool = True, **kwargs) → requests.models.Response

Send a HTTP request to the ACS.

Parameters:
  • data – The request body
  • fix_cwmp_id – If true, the cwmp:ID in the body will replaced with the cwmp:ID in the last response.
  • **kwargs – Arguments passed to self._session.post()
Returns:

The ACS’ response.
replay(request: typing.Union[requests.models.PreparedRequest, NoneType] = None) → requests.models.Response

Replay a request that has previously been sent. Useful to e.g. test for nonce re-use.

Parameters:request – The request to replay. If no request is passed, the last request will be replayed.
done() → requests.models.Response

Indicate to the ACS that the client has finished sending RPCs

handle_server_rpcs() → int

Handle server RPCs automatically, starting with the last already transmitted RPC. This is usually called immediately after .done()

Returns:The number of handled RPCs
Raises:NotImplementedError if automated handling of the RPC is not implemented.
close() → None

Close any existing connections to the ACS and reset the session.

Remote Procedure Calls

For method signatures, see the RPC section below.

class tr069.Client
inform(**kwargs) → requests.models.Response

Create a Inform RPC

get_rpc_methods() → requests.models.Response

Create a GetRPCMethods RPC

request_download(*args, **kwargs) → requests.models.Response

Create a RequestDownload RPC

set_parameter_values_response(*args, **kwargs) → requests.models.Response

Create a SetParameterValuesResponse

get_parameter_values_response(*args, **kwargs) → requests.models.Response

Make a GetParameterValuesResponse

set_parameter_attributes_response() → requests.models.Response

Make a SetParameterAttributesResponse

get_parameter_names_response(*args, **kwargs) → requests.models.Response

Make a GetParameterNamesResponse

Connection Request Server

class tr069.ConnectionRequestServer(address: typing.Tuple[str, int] = ('', 7547), handle_manually: bool = False) → None

A minimal TR-069 Connection Request Server.

Main purpose of this module is to detect that providers are sending connection requests, it implements no authentication whatsoever. Connection requests are acknowledged automatically by default and will be printed to stdout. For non-interactive use, users may call instance.queue.get() to wait for a connection request.

Proxy Support

tr069.proxy.enable(url: str, ca_certs: typing.Union[str, NoneType] = None) → None

Route all outgoing requests through a proxy.

Parameters:
  • url – The proxy URL.
  • ca_certs – path to a certificate list stored in PEM format (optional).
tr069.proxy.disable() → None

Disable proxying of requests.

Devices

class tr069.data.device.Device(manufacturer, oui, product_class, serial, params: typing.Union[tr069.data.parameters.Parameters, NoneType] = None)

A TR-069 Device.

device.from_xml(xml: str) → tr069.data.device.Device

Construct a device from an Inform XML.

tr069.data.device.AVM_FRITZ_BOX_7490

Device(AVM FRITZ!Box, Serial – 0896D776FAA2, 11 parameters)

tr069.data.device.FREECWMP

Device(freecwmp freecwmp, Serial – FFFFFF123456, 11 parameters)

tr069.data.device.DEFAULT

Device(AVM FRITZ!Box, Serial – 0896D776FAA2, 11 parameters)

Inform Events

class tr069.data.event.Event(code, command_key='')

A TR-069 EventStruct. Represents an event that caused a TR-069 connection.

tr069.data.event.Bootstrap

Event(0 BOOTSTRAP)

tr069.data.event.Boot

Event(1 Boot)

tr069.data.event.Periodic

Event(2 PERIODIC)

tr069.data.event.Scheduled

Event(3 SCHEDULED)

tr069.data.event.ValueChange

Event(4 VALUE CHANGE)

tr069.data.event.Kicked

Event(5 KICKED)

tr069.data.event.ConnectionRequest

Event(6 CONNECTION REQUEST)

tr069.data.event.TransferComplete

Event(7 TRANSFER COMPLETE)

tr069.data.event.DiagnosticsComplete

Event(8 DIAGNOSTICS COMPLETE)

tr069.data.event.RequestDownload

Event(9 REQUEST DOWNLOAD)

tr069.data.event.AutonomousTransferComplete

Event(10 AUTONOMOUS TRANSFER COMPLETE)

tr069.data.event.DuStateChangeComplete

Event(11 DU STATE CHANGE COMPLETE)

tr069.data.event.AutonomousDuStateChangeComplete

Event(12 AUTONOMOUS DU STATE CHANGE COMPLETE)

tr069.data.event.Wakeup

Event(13 WAKEUP)

Device Parameters

parameters.from_xml(xml: str) → tr069.data.parameters.Parameters

Construct a parameters object from XML that contains a ParameterValueStruct, e.g., an Inform RPC.

class tr069.data.parameters.Parameter(name, value, type='xsd:string', notification_level: int = 0, writable: bool = True)

A TR-069 ParameterValueStruct. Represents a parameter on the device as specified by the TR-069 data models.

class tr069.data.parameters.Parameters(params: typing.Iterable[tr069.data.parameters.Parameter] = (), **kwargs: typing.Dict[str, str])

A collection of TR-069 parameters representing a device.

all(key: str = '', min_notification_level: int = 0) → typing.List[tr069.data.parameters.Parameter]
Returns:
A list containing...
  • all parameter starting with the common prefix key, if key is empty or ends with a dot.
  • the requested parameter key.
tr069.data.parameters.from_xml(xml: str) → tr069.data.parameters.Parameters

Construct a parameters object from XML that contains a ParameterValueStruct, e.g., an Inform RPC.

Remote Procedure Calls

tr069.data.rpcs.make_get_rpc_methods() → str

Create a GetRPCMethods RPC

tr069.data.rpcs.make_inform(*, device: tr069.data.device.Device, events: typing.Collection[tr069.data.event.Event] = (Event(4 VALUE CHANGE), Event(1 Boot), Event(0 BOOTSTRAP)), params: typing.Union[typing.Collection[tr069.data.parameters.Parameter], NoneType] = None, time: typing.Union[datetime.datetime, NoneType] = None, retry_count: int = 0) → str

Create a Inform RPC

tr069.data.rpcs.make_set_parameter_values_response(status: int = 0) → str

Create a SetParameterValuesResponse

tr069.data.rpcs.parse_set_parameter_values(xml: str) → typing.List[tr069.data.parameters.Parameter]

Parse a SetParameterValues RPC

Returns:List of parameters that should be updated.
tr069.data.rpcs.make_get_parameter_values_response(params: typing.Collection[tr069.data.parameters.Parameter] = ()) → str

Make a GetParameterValuesResponse

tr069.data.rpcs.parse_get_parameter_values(xml: str) → typing.List[str]

Parse a GetParameterValues RPC.

Returns:List of requested parameters.
tr069.data.rpcs.make_get_parameter_names_response(params: typing.Collection[tr069.data.parameters.Parameter] = ()) → str

Make a GetParameterNamesResponse

tr069.data.rpcs.parse_get_parameter_names(xml: str) → typing.Tuple[str, bool]

Parse a GetParameterNames RPC.

Returns:A (path, next_level) tuple.
tr069.data.rpcs.make_set_parameter_attributes_response() → str

Make a SetParameterAttributesResponse

tr069.data.rpcs.make_request_download(file_type: str = '1 Firmware Upgrade Image', file_type_arg=None) → str

Create a RequestDownload RPC

requests.Response

class requests.models.Response

The Response object, which contains a server’s response to an HTTP request.

status_code = None

Integer Code of responded HTTP Status, e.g. 404 or 200.

headers = None

Case-insensitive Dictionary of Response Headers. For example, headers['content-encoding'] will return the value of a 'Content-Encoding' response header.

raw = None

File-like object representation of response (for advanced usage). Use of raw requires that stream=True be set on the request.

url = None

Final URL location of Response.

encoding = None

Encoding to decode with when accessing r.text.

history = None

A list of Response objects from the history of the Request. Any redirect responses will end up here. The list is sorted from the oldest to the most recent request.

reason = None

Textual reason of responded HTTP Status, e.g. “Not Found” or “OK”.

cookies = None

A CookieJar of Cookies the server sent back.

elapsed = None

The amount of time elapsed between sending the request and the arrival of the response (as a timedelta). This property specifically measures the time taken between sending the first byte of the request and finishing parsing the headers. It is therefore unaffected by consuming the response content or the value of the stream keyword argument.

request = None

The PreparedRequest object to which this is a response.

is_redirect

True if this Response is a well-formed HTTP redirect that could have been processed automatically (by Session.resolve_redirects()).

is_permanent_redirect

True if this Response one of the permanent versions of redirect

apparent_encoding

The apparent encoding, provided by the chardet library

iter_content(chunk_size=1, decode_unicode=False)

Iterates over the response data. When stream=True is set on the request, this avoids reading the content at once into memory for large responses. The chunk size is the number of bytes it should read into memory. This is not necessarily the length of each item returned as decoding can take place.

chunk_size must be of type int or None. A value of None will function differently depending on the value of stream. stream=True will read data as it arrives in whatever size the chunks are received. If stream=False, data is returned as a single chunk.

If decode_unicode is True, content will be decoded using the best available encoding based on the response.

iter_lines(chunk_size=512, decode_unicode=None, delimiter=None)

Iterates over the response data, one line at a time. When stream=True is set on the request, this avoids reading the content at once into memory for large responses.

Note

This method is not reentrant safe.

content

Content of the response, in bytes.

text

Content of the response, in unicode.

If Response.encoding is None, encoding will be guessed using chardet.

The encoding of the response content is determined based solely on HTTP headers, following RFC 2616 to the letter. If you can take advantage of non-HTTP knowledge to make a better guess at the encoding, you should set r.encoding appropriately before accessing this property.

json(**kwargs)

Returns the json-encoded content of a response, if any.

Parameters:**kwargs – Optional arguments that json.loads takes.
Raises:ValueError – If the response body does not contain valid json.

Returns the parsed header links of the response, if any.

raise_for_status()

Raises stored HTTPError, if one occurred.

close()

Releases the connection back to the pool. Once this method has been called the underlying raw object must not be accessed again.

Note: Should not normally need to be called explicitly.