sane_ssdp: cache the UPNP root device
i could also cache the lan and wan... future work, i guess
This commit is contained in:
parent
8bd7fa8a3f
commit
307121ec2c
|
@ -31,6 +31,14 @@ class SsdpResponse:
|
||||||
def location(self) -> str:
|
def location(self) -> str:
|
||||||
return self.headers.get("LOCATION")
|
return self.headers.get("LOCATION")
|
||||||
|
|
||||||
|
def get_cached_root_devices() -> list[str]:
|
||||||
|
try:
|
||||||
|
dev = open("/var/lib/uninsane/upnp.txt", "r").read()
|
||||||
|
except IOError:
|
||||||
|
return []
|
||||||
|
else:
|
||||||
|
logger.debug("loaded cached UPNP root device", dev)
|
||||||
|
return [dev]
|
||||||
|
|
||||||
def get_root_devices() -> list[str]:
|
def get_root_devices() -> list[str]:
|
||||||
listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
||||||
|
@ -91,12 +99,15 @@ def get_ips_from_location(location: str) -> tuple[str | None, str | None]:
|
||||||
logger.info(f"got LAN = {lan} from {location}")
|
logger.info(f"got LAN = {lan} from {location}")
|
||||||
return lan, wan
|
return lan, wan
|
||||||
|
|
||||||
def get_any_wan() -> tuple[str, str, str] | None:
|
def get_any_wan(cached: bool = False) -> tuple[str, str, str] | None:
|
||||||
""" return (location, LAN IP, WAN IP) for the first device seen which has a WAN IP """
|
""" return (location, LAN IP, WAN IP) for the first device seen which has a WAN IP """
|
||||||
for location in get_root_devices():
|
sources = ([ get_cached_root_devices() ] if cached else []) \
|
||||||
lan, wan = get_ips_from_location(location)
|
+ [ get_root_devices() ]
|
||||||
if lan and wan:
|
for source in sources:
|
||||||
return location, lan, wan
|
for location in source:
|
||||||
|
lan, wan = get_ips_from_location(location)
|
||||||
|
if lan and wan:
|
||||||
|
return location, lan, wan
|
||||||
|
|
||||||
def forward_port(root_device: str, proto: str, port: int, lan_ip: str, reason: str = "", duration: int = 86400) -> None:
|
def forward_port(root_device: str, proto: str, port: int, lan_ip: str, reason: str = "", duration: int = 86400) -> None:
|
||||||
args = [
|
args = [
|
||||||
|
|
|
@ -99,6 +99,6 @@ if __name__ == '__main__':
|
||||||
print(e)
|
print(e)
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
root_device, lan, _wan = get_any_wan()
|
root_device, lan, _wan = get_any_wan(cached=True)
|
||||||
for (proto, port, reason) in forwards:
|
for (proto, port, reason) in forwards:
|
||||||
forward_port(root_device, proto, port, lan_ip=lan, reason=reason, duration=duration)
|
forward_port(root_device, proto, port, lan_ip=lan, reason=reason, duration=duration)
|
||||||
|
|
Loading…
Reference in New Issue
Block a user