LBRYCommunity/aioupnp
1
0
Fork 0
Code Issues 7 Pull requests 1 Projects Releases 1 Packages Wiki Activity

Merge pull request #24 from lbryio/debug-gateway

Browse source 
add `gather_debug_info` command
This commit is contained in:
Jack Robison 2020-10-13 15:57:16 -04:00 committed by GitHub
commit 33bfa23273
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 96 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

62
README.md
View file

@ -39,42 +39,64 @@ aioupnp [-h] [--debug_logging] [--interface=<interface>] [--gateway_address=<gat
[--lan_address=<lan_address>] [--timeout=<timeout>]
[(--<case sensitive m-search header>=<value>)...]
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
#### 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
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
#### 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
#### 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
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.
## License

80
aioupnp/upnp.py
View file

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

2
tests/test_cli.py
View file

@ -48,7 +48,7 @@ in the order they are to be used. For example:
Commands:
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>
"""