StevenLooman / async_upnp_client

Async UPnP Client for Python
Other
46 stars 37 forks source link

Can this module add/remove port mapping #219

Closed brostosjoined closed 8 months ago

StevenLooman commented 8 months ago

Hi @brostosjoined, thank you for this issue. Yes, it is possible to add/remove port mappings using this library. If you wish to do to add a port mapping from within Python you can use https://github.com/StevenLooman/async_upnp_client/blob/858cb4cc1b3947499742735e15329ba2f1c12ce8/async_upnp_client/profiles/igd.py#L383. To delete a port mapping you can use: https://github.com/StevenLooman/async_upnp_client/blob/858cb4cc1b3947499742735e15329ba2f1c12ce8/async_upnp_client/profiles/igd.py#L425

Does this answer your question?

brostosjoined commented 8 months ago

Great let me test it if it works ill make an example

brostosjoined commented 8 months ago

@StevenLooman can you provide a simple example on how to search for the upnp device on the network and pass it to the functions you provided above

StevenLooman commented 8 months ago

What exactly are you looking for? Just a means to add a port mapping, or specifically from within python? If the former, then perhaps the upnp-client CLI tool is usable for you, see: https://github.com/StevenLooman/async_upnp_client?tab=readme-ov-file#upnp-client, specifically upnp-client search will most likely provide the information you are looking for.

If the latter, you can use this to search: https://github.com/StevenLooman/async_upnp_client/blob/858cb4cc1b3947499742735e15329ba2f1c12ce8/async_upnp_client/search.py#L162 But the CLI will most likely also show you how to search. After you've gotten the search results, you'll have to construct a UpnpDevice using the UpnpFactory, and then instantiate a new IgdDevice.

brostosjoined commented 8 months ago

What exactly are you looking for? Just a means to add a port mapping, or specifically from within python? If the former, then perhaps the upnp-client CLI tool is usable for you, see: https://github.com/StevenLooman/async_upnp_client?tab=readme-ov-file#upnp-client, specifically upnp-client search will most likely provide the information you are looking for.

If the latter, you can use this to search:

https://github.com/StevenLooman/async_upnp_client/blob/858cb4cc1b3947499742735e15329ba2f1c12ce8/async_upnp_client/search.py#L162

But the CLI will most likely also show you how to search. After you've gotten the search results, you'll have to construct a UpnpDevice using the UpnpFactory, and then instantiate a new IgdDevice.

As you mentioned here above I was trying that maybe ended up sending a wrong value to the to the command line now it doesnt work and it was previously working perfectly

upnp-client --debug --pprint search --search_target "ssdp:all"

now it doesnt work at all can you check on that here is the debug output

 upnp-client --debug --pprint search
DEBUG:async_upnp_client.search:Start listening for search responses
DEBUG:async_upnp_client.ssdp:Creating socket, source: (<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_DGRAM: 2>, 17, '', ('0.0.0.0', 0)), target: (<AddressFamily.AF_INET: 2>, <SocketKind.SOCK_DGRAM: 2>, 17, '', ('239.255.255.250', 1900))
DEBUG:async_upnp_client.search:Binding socket, socket: <socket.socket fd=764, family=2, type=2, proto=0>, address: ('0.0.0.0', 0)
DEBUG:async_upnp_client.ssdp:Connection made, transport: <_ProactorDatagramTransport fd=764>, socket: <asyncio.TransportSocket fd=764, family=2, type=2, proto=0, laddr=('0.0.0.0', 58907)>
DEBUG:async_upnp_client.search:On connect, transport: <_ProactorDatagramTransport fd=764>, socket: <asyncio.TransportSocket fd=764, family=2, type=2, proto=0, laddr=('0.0.0.0', 58907)>
DEBUG:async_upnp_client.search:Sending SEARCH packet, transport: <_ProactorDatagramTransport fd=764 read=<_OverlappedFuture pending cb=[_ProactorDatagramTransport._loop_reading()]>>, socket: <asyncio.TransportSocket fd=764, family=2, type=2, proto=0, laddr=('0.0.0.0', 58907)>, override_target: None
DEBUG:async_upnp_client.ssdp:Sending SSDP packet, transport: <_ProactorDatagramTransport fd=764 read=<_OverlappedFuture pending cb=[_ProactorDatagramTransport._loop_reading()]>>, socket: <asyncio.TransportSocket fd=764, family=2, type=2, proto=0, laddr=('0.0.0.0', 58907)>, target: ('239.255.255.250', 1900)
DEBUG:async_upnp_client.ssdp:Lost connection, error: None, transport: <_ProactorDatagramTransport closing fd=764>, socket: <asyncio.TransportSocket fd=764, family=2, type=2, proto=0, laddr=('0.0.0.0', 58907)>
StevenLooman commented 8 months ago

You can omit the search_target option for upnp-client. It searches for all devices/services by default. I do get results when running that command, however. Perhaps a firewall is blocking traffic? You can also try binding to a specific IP/network adapter using the --bind option.

Regarding an example program, I think (untested/gotten+adapted from somewhere else), this is something you are looking for:

#!/usr/bin/env python3
from async_upnp_client.aiohttp import AiohttpRequester
from async_upnp_client.client import UpnpDevice
from async_upnp_client.client_factory import UpnpFactory
from async_upnp_client.profiles.igd import IgdDevice
from async_upnp_client.search import async_search
from async_upnp_client.utils import get_local_ip

async def build_device(search_target: str) -> UpnpDevice:
    """Find and construct device."""
    location: Optional[str] = None
    async def callback(headers) -> None:
        """Search callback."""
        nonlocal location
        location = headers['location']

    # Do the search, this blocks for timeout (4 seconds, default).
    await async_search(callback, search_target=search_target)

    if location:
        requester = AiohttpRequester()
        factory = UpnpFactory(requester, non_strict=True)
        device = await factory.async_create_device(description_url=location)
        return device

async def async_main() -> None:
    """Main."""
    device = await build_device(search_target=search_target)
    if not device:
        print("Could not find device")
        sys.exit(1)

    igd_device = IgdDevice(device, None)
    await igd_device.async_add_port_mapping(...)  # <-- Fill in the arguments.

def main() -> None:
    try:
        asyncio.run(async_main())
    except KeyboardInterrupt:
        pass

if __name__ == '__main__':
    main()

Hope this helps.

brostosjoined commented 8 months ago

220 this closes this