1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
#!/usr/bin/env python3
"""Example of adding and deleting a port mapping."""
import asyncio
import ipaddress
import sys
from datetime import timedelta
from typing import cast
from async_upnp_client.aiohttp import AiohttpRequester
from async_upnp_client.client_factory import UpnpFactory
from async_upnp_client.profiles.igd import IgdDevice
from async_upnp_client.utils import CaseInsensitiveDict, get_local_ip
SOURCE = ("0.0.0.0", 0)
async def discover_igd_devices() -> set[CaseInsensitiveDict]:
"""Discover IGD devices."""
# Do the search, this blocks for timeout (4 seconds, default).
discoveries = await IgdDevice.async_search(source=SOURCE)
if not discoveries:
print("Could not find device")
sys.exit(1)
return discoveries
async def build_igd_device(discovery: CaseInsensitiveDict) -> IgdDevice:
"""Find and construct device."""
location = discovery["location"]
requester = AiohttpRequester()
factory = UpnpFactory(requester, non_strict=True)
device = await factory.async_create_device(description_url=location)
return IgdDevice(device, None)
async def async_add_port_mapping(igd_device: IgdDevice) -> None:
"""Add port mapping."""
external_ip_address = await igd_device.async_get_external_ip_address()
if not external_ip_address:
print("Could not get external IP address")
sys.exit(1)
remote_host = ipaddress.ip_address(external_ip_address)
remote_host_ipv4 = cast(ipaddress.IPv4Address, remote_host)
local_ip = ipaddress.ip_address(get_local_ip())
local_ip_ipv4 = cast(ipaddress.IPv4Address, local_ip)
# Change `enabled` to False to disable port mapping.
# NB: This does not delete the port mapping.
enabled = True
mapping_name = "Bombsquad"
await igd_device.async_add_port_mapping(
remote_host=remote_host_ipv4,
external_port=43210,
internal_client=local_ip_ipv4,
internal_port=43210,
protocol="UDP",
enabled=enabled,
description=mapping_name,
lease_duration=timedelta(seconds=7200),
) # Time in secs
async def async_del_port_mapping(igd_device: IgdDevice) -> None:
"""Delete port mapping."""
external_ip_address = await igd_device.async_get_external_ip_address()
if not external_ip_address:
print("Could not get external IP address")
sys.exit(1)
remote_host = ipaddress.ip_address(external_ip_address)
remote_host_ipv4 = cast(ipaddress.IPv4Address, remote_host)
await igd_device.async_delete_port_mapping(
remote_host=remote_host_ipv4,
external_port=43210,
protocol="UDP",
)
async def async_main() -> None:
"""Async main."""
discoveries = await discover_igd_devices()
print(f"Discoveries: {discoveries}")
discovery = list(discoveries)[0]
print(f"Using device at location: {discovery['location']}")
igd_device = await build_igd_device(discovery)
print("Creating port mapping")
await async_add_port_mapping(igd_device)
await asyncio.sleep(5)
print("Deleting port mapping")
await async_del_port_mapping(igd_device)
def main() -> None:
"""Main."""
try:
asyncio.run(async_main())
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()
|
