-
Notifications
You must be signed in to change notification settings - Fork 228
Expand file tree
/
Copy pathasync_apple_scanner.py
More file actions
executable file
·133 lines (112 loc) · 4.5 KB
/
async_apple_scanner.py
File metadata and controls
executable file
·133 lines (112 loc) · 4.5 KB
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
#!/usr/bin/env python
"""Scan for apple devices."""
from __future__ import annotations
import argparse
import asyncio
import logging
from typing import Any, cast
from zeroconf import DNSQuestionType, IPVersion, ServiceStateChange, Zeroconf
from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf
HOMESHARING_SERVICE: str = "_appletv-v2._tcp.local."
DEVICE_SERVICE: str = "_touch-able._tcp.local."
MEDIAREMOTE_SERVICE: str = "_mediaremotetv._tcp.local."
AIRPLAY_SERVICE: str = "_airplay._tcp.local."
COMPANION_SERVICE: str = "_companion-link._tcp.local."
RAOP_SERVICE: str = "_raop._tcp.local."
AIRPORT_ADMIN_SERVICE: str = "_airport._tcp.local."
DEVICE_INFO_SERVICE: str = "_device-info._tcp.local."
ALL_SERVICES = [
HOMESHARING_SERVICE,
DEVICE_SERVICE,
MEDIAREMOTE_SERVICE,
AIRPLAY_SERVICE,
COMPANION_SERVICE,
RAOP_SERVICE,
AIRPORT_ADMIN_SERVICE,
DEVICE_INFO_SERVICE,
]
log = logging.getLogger(__name__)
_PENDING_TASKS: set[asyncio.Task] = set()
def async_on_service_state_change(
zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange
) -> None:
print(f"Service {name} of type {service_type} state changed: {state_change}")
if state_change is not ServiceStateChange.Added:
return
base_name = name[: -len(service_type) - 1]
device_name = f"{base_name}.{DEVICE_INFO_SERVICE}"
task = asyncio.ensure_future(_async_show_service_info(zeroconf, service_type, name))
_PENDING_TASKS.add(task)
task.add_done_callback(_PENDING_TASKS.discard)
# Also probe for device info
task = asyncio.ensure_future(_async_show_service_info(zeroconf, DEVICE_INFO_SERVICE, device_name))
_PENDING_TASKS.add(task)
task.add_done_callback(_PENDING_TASKS.discard)
async def _async_show_service_info(zeroconf: Zeroconf, service_type: str, name: str) -> None:
info = AsyncServiceInfo(service_type, name)
await info.async_request(zeroconf, 3000, question_type=DNSQuestionType.QU)
print(f"Info from zeroconf.get_service_info: {info!r}")
if info:
addresses = [f"{addr}:{cast(int, info.port)}" for addr in info.parsed_addresses()]
print(f" Name: {name}")
print(f" Addresses: {', '.join(addresses)}")
print(f" Weight: {info.weight}, priority: {info.priority}")
print(f" Server: {info.server}")
if info.properties:
print(" Properties are:")
for key, value in info.properties.items():
print(f" {key!r}: {value!r}")
else:
print(" No properties")
else:
print(" No info")
print("\n")
class AsyncAppleScanner:
def __init__(self, args: Any) -> None:
self.args = args
self.aiobrowser: AsyncServiceBrowser | None = None
self.aiozc: AsyncZeroconf | None = None
async def async_run(self) -> None:
self.aiozc = AsyncZeroconf(ip_version=ip_version)
await self.aiozc.zeroconf.async_wait_for_start()
print(f"\nBrowsing {ALL_SERVICES} service(s), press Ctrl-C to exit...\n")
kwargs = {
"handlers": [async_on_service_state_change],
"question_type": DNSQuestionType.QU,
}
if self.args.target:
kwargs["addr"] = self.args.target
self.aiobrowser = AsyncServiceBrowser(
self.aiozc.zeroconf,
ALL_SERVICES,
**kwargs, # type: ignore[arg-type]
)
await asyncio.Event().wait()
async def async_close(self) -> None:
assert self.aiozc is not None
assert self.aiobrowser is not None
await self.aiobrowser.async_cancel()
await self.aiozc.async_close()
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
parser = argparse.ArgumentParser()
parser.add_argument("--debug", action="store_true")
version_group = parser.add_mutually_exclusive_group()
version_group.add_argument("--target", help="Unicast target")
version_group.add_argument("--v6", action="store_true")
version_group.add_argument("--v6-only", action="store_true")
args = parser.parse_args()
if args.debug:
logging.getLogger("zeroconf").setLevel(logging.DEBUG)
if args.v6:
ip_version = IPVersion.All
elif args.v6_only:
ip_version = IPVersion.V6Only
else:
ip_version = IPVersion.V4Only
loop = asyncio.get_event_loop()
runner = AsyncAppleScanner(args)
try:
loop.run_until_complete(runner.async_run())
except KeyboardInterrupt:
loop.run_until_complete(runner.async_close())