add gather_debug_info command #24

Merged
jackrobison merged 2 commits from debug-gateway into master 2020-10-13 21:57:17 +02:00
4 changed files with 96 additions and 54 deletions

View file

@ -39,42 +39,64 @@ aioupnp [-h] [--debug_logging] [--interface=<interface>] [--gateway_address=<gat
[--lan_address=<lan_address>] [--timeout=<timeout>] [--lan_address=<lan_address>] [--timeout=<timeout>]
[(--<case sensitive m-search header>=<value>)...] [(--<case sensitive m-search header>=<value>)...]
command [--<arg name>=<arg>]... command [--<arg name>=<arg>]...
If m-search headers are provided as keyword arguments all of the headers to be used must be provided,
in the order they are to be used. For example:
aioupnp --HOST=239.255.255.250:1900 --MAN=\"ssdp:discover\" --MX=1 --ST=upnp:rootdevice m_search
``` ```
### Commands
m_search | add_port_mapping | get_port_mapping_by_index | get_redirects | get_specific_port_mapping | delete_port_mapping | get_next_mapping
#### Commands
* `help`
* `get_external_ip`
* `m_search`
* `add_port_mapping`
* `get_port_mapping_by_index`
* `get_redirects`
* `get_specific_port_mapping`
* `delete_port_mapping`
* `get_next_mapping`
* `gather_debug_info`
### Examples #### To get the documentation for a command
#### To get the external ip address from the UPnP gateway aioupnp help get_external_ip
#### To get the external ip address
aioupnp get_external_ip aioupnp get_external_ip
#### To set up a TCP port redirect
aioupnp add_port_mapping --external_port=1234 --internal_port=1234 --lan_address=<lan_addr> --description=test --protocol=TCP
#### To list the active port mappings on the gateway #### To list the active port mappings on the gateway
aioupnp get_redirects aioupnp get_redirects
#### To debug the gateway discovery #### To set up a TCP port mapping
aioupnp add_port_mapping --external_port=1234 --internal_port=1234 --lan_address=<lan_addr> --description=test --protocol=TCP
#### To delete a TCP port mapping
aioupnp delete_port_mapping --external_port=1234 --protocol=TCP
#### M-Search headers
UPnP uses multicast protocol - SSDP - to locate the gateway. Gateway discovery is automatic by default, but you may provide specific headers for the search to use to override automatic discovery.
If m-search headers are provided as keyword arguments then all of the headers to be used must be provided, in the order they are to be used. For example:
aioupnp --HOST=239.255.255.250:1900 --MAN=\"ssdp:discover\" --MX=1 --ST=upnp:rootdevice m_search
#### Using non-default network interfaces
By default, the network device will be automatically discovered. The interface may instead be specified with the `--interface`, provided before the command to be run. The gateway used on the interface network may be specified with the `--gateway_address` argument.
aioupnp --interface=wlp4s0 --gateway_address=192.168.1.6 m_search
## Troubleshooting
#### Debug logging
To enable verbose debug logging, add the `--debug_logging` argument before the command
aioupnp --debug_logging m_search aioupnp --debug_logging m_search
#### To debug a gateway on a non default network interface #### It really doesn't work
If aioupnp doesn't work with a device, a debugging report can be collected with `aioupnp gather_debug_info`.
aioupnp --interface=vmnet1 --debug_logging m_search This will attempt to discover the UPnP gateway, and then perform a functionality check where it will request the external address and existing port mappings before attempting to make and remove a port mapping. The final result is the zipped packet dump of these attempts, which allows writing tests replaying it.
#### To debug a gateway on a non default network interface that isn't the router
aioupnp --interface=vmnet1 --gateway_address=192.168.1.106 --debug_logging m_search
## License ## License

View file

@ -1,6 +1,5 @@
# import os import zlib
# import zlib import base64
# import base64
import logging import logging
import json import json
import asyncio import asyncio
@ -8,21 +7,12 @@ from typing import Tuple, Dict, List, Union, Optional, Any
from aioupnp.fault import UPnPError from aioupnp.fault import UPnPError
from aioupnp.gateway import Gateway from aioupnp.gateway import Gateway
from aioupnp.interfaces import get_gateway_and_lan_addresses from aioupnp.interfaces import get_gateway_and_lan_addresses
from aioupnp.serialization.ssdp import SSDPDatagram
from aioupnp.commands import GetGenericPortMappingEntryResponse, GetSpecificPortMappingEntryResponse from aioupnp.commands import GetGenericPortMappingEntryResponse, GetSpecificPortMappingEntryResponse
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
# def _encode(x):
# if isinstance(x, bytes):
# return x.decode()
# elif isinstance(x, Exception):
# return str(x)
# return x
class UPnP: class UPnP:
def __init__(self, lan_address: str, gateway_address: str, gateway: Gateway) -> None: def __init__(self, lan_address: str, gateway_address: str, gateway: Gateway) -> None:
self.lan_address = lan_address self.lan_address = lan_address
@ -248,22 +238,52 @@ class UPnP:
await self.add_port_mapping(port, protocol, _internal_port, self.lan_address, description) await self.add_port_mapping(port, protocol, _internal_port, self.lan_address, description)
return port return port
# @cli async def gather_debug_info(self) -> str: # pragma: no cover
# async def debug_gateway(self) -> str: """
# return json.dumps({ Gather debugging information for this gateway, used for generating test cases for devices with errors.
# "gateway": self.gateway.debug_gateway(),
# "client_address": self.lan_address, :return: (str) compressed debugging information
# }, default=_encode, indent=2) """
#
# @property def _encode(x):
# def zipped_debugging_info(self) -> str: if isinstance(x, bytes):
# return base64.b64encode(zlib.compress( return x.decode()
# json.dumps({ elif isinstance(x, Exception):
# "gateway": self.gateway.debug_gateway(), return str(x)
# "client_address": self.lan_address, return x
# }, default=_encode, indent=2).encode()
# )).decode() try:
# await self.get_external_ip()
except UPnPError:
pass
try:
await self.get_redirects()
except UPnPError:
pass
try:
external_port = await self.get_next_mapping(1234, 'TCP', 'aioupnp testing')
except UPnPError:
external_port = None
try:
await self.get_redirects()
except UPnPError:
pass
if external_port:
try:
await self.delete_port_mapping(external_port, 'TCP')
except UPnPError:
pass
try:
await self.get_redirects()
except UPnPError:
pass
return base64.b64encode(zlib.compress(
json.dumps({
"gateway": self.gateway.debug_gateway(),
"client_address": self.lan_address,
}, default=_encode, indent=2).encode()
)).decode()
# @cli # @cli
# async def get_natrsip_status(self) -> Tuple[bool, bool]: # async def get_natrsip_status(self) -> Tuple[bool, bool]:
# """Returns (NewRSIPAvailable, NewNATEnabled)""" # """Returns (NewRSIPAvailable, NewNATEnabled)"""
@ -362,7 +382,8 @@ cli_commands = [
'get_redirects', 'get_redirects',
'get_specific_port_mapping', 'get_specific_port_mapping',
'delete_port_mapping', 'delete_port_mapping',
'get_next_mapping' 'get_next_mapping',
'gather_debug_info'
] ]
@ -372,7 +393,6 @@ def run_cli(method: str, igd_args: Dict[str, Union[bool, str, int]], lan_address
loop: Optional[asyncio.AbstractEventLoop] = None) -> None: loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
kwargs = kwargs or {} kwargs = kwargs or {}
igd_args = igd_args
timeout = int(timeout) timeout = int(timeout)
loop = loop or asyncio.get_event_loop() loop = loop or asyncio.get_event_loop()
fut: 'asyncio.Future' = loop.create_future() fut: 'asyncio.Future' = loop.create_future()

View file

@ -48,7 +48,7 @@ in the order they are to be used. For example:
Commands: Commands:
m_search | get_external_ip | add_port_mapping | get_port_mapping_by_index | get_redirects | m_search | get_external_ip | add_port_mapping | get_port_mapping_by_index | get_redirects |
get_specific_port_mapping | delete_port_mapping | get_next_mapping get_specific_port_mapping | delete_port_mapping | get_next_mapping | gather_debug_info
For help with a specific command: aioupnp help <command> For help with a specific command: aioupnp help <command>
""" """