Files
Android-pyApi/android_pyapi/android/net.py
T
shuiyu1123 bfe27fdc08 feat: Android-pyApi v1.0 - 全面翻译 Android SDK API 为 Python
52 个模块覆盖 Android + Java + org.json 标准库
2026-06-24 11:16:16 +00:00

1397 lines
46 KiB
Python

"""
Android Network 组件 — ConnectivityManager, NetworkInfo, Uri_Builder 等
方法签名完全对齐 Android SDK
"""
import os
import sys
import socket
import struct
import threading
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
from urllib.parse import urlparse, urlunparse, quote, unquote
from .._core import (
Stub, AndroidOnlyError, android_context,
resolve_android_path, to_android_path,
_sdcard_root, _app_data_dir, _package_name,
)
# ═══════════════════════════════════════════════════════════════
# Uri (re-export from content)
# ═══════════════════════════════════════════════════════════════
# Uri is defined in android.content; re-export the reference here.
from .content import Uri
Uri_Builder = Uri.Builder
# ═══════════════════════════════════════════════════════════════
# LocalSocket / LocalSocketAddress
# ═══════════════════════════════════════════════════════════════
class LocalSocketAddress(Stub):
"""Android LocalSocketAddress"""
class Namespace(Stub):
ABSTRACT = 0
RESERVED = 1
FILESYSTEM = 2
def __init__(self, name: str, namespace: int = Namespace.ABSTRACT):
super().__init__()
self._name = name
self._namespace = namespace
def getName(self) -> str:
return self._name
def getNamespace(self) -> int:
return self._namespace
class LocalSocket(Stub):
"""Android LocalSocket — 基于 Unix domain socket 实现"""
def __init__(self, sock_type: int = 1): # SOCK_STREAM
super().__init__()
self._sock: Optional[socket.socket] = None
self._sock_type = sock_type
self._input_stream: Optional[bytes] = b""
self._output_stream: Optional[bytes] = b""
self._closed = False
self._so_timeout: float = 0.0
self._address: Optional[LocalSocketAddress] = None
def connect(self, address: LocalSocketAddress, timeout: int = 0):
self._address = address
ns = address.getNamespace()
if ns == LocalSocketAddress.Namespace.ABSTRACT:
sock_path = "\0" + address.getName()
elif ns == LocalSocketAddress.Namespace.FILESYSTEM:
sock_path = address.getName()
else:
sock_path = os.path.join("/dev/socket", address.getName())
try:
self._sock = socket.socket(socket.AF_UNIX, self._sock_type)
if timeout > 0:
self._sock.settimeout(timeout / 1000.0)
self._sock.connect(sock_path)
except (OSError, socket.error) as e:
raise AndroidOnlyError(f"LocalSocket.connect failed: {e}")
def bind(self, address: LocalSocketAddress):
self._address = address
ns = address.getNamespace()
if ns == LocalSocketAddress.Namespace.ABSTRACT:
sock_path = "\0" + address.getName()
elif ns == LocalSocketAddress.Namespace.FILESYSTEM:
sock_path = address.getName()
else:
sock_path = os.path.join("/dev/socket", address.getName())
try:
self._sock = socket.socket(socket.AF_UNIX, self._sock_type)
self._sock.bind(sock_path)
except (OSError, socket.error) as e:
raise AndroidOnlyError(f"LocalSocket.bind failed: {e}")
def getInputStream(self):
return self._input_stream
def getOutputStream(self):
return self._output_stream
def close(self):
self._closed = True
if self._sock:
try:
self._sock.close()
except Exception:
pass
self._sock = None
def setSoTimeout(self, timeout: int):
self._so_timeout = timeout / 1000.0
if self._sock:
self._sock.settimeout(self._so_timeout)
def getSoTimeout(self) -> int:
return int(self._so_timeout * 1000)
def shutdownInput(self):
if self._sock:
try:
self._sock.shutdown(socket.SHUT_RD)
except Exception:
pass
def shutdownOutput(self):
if self._sock:
try:
self._sock.shutdown(socket.SHUT_WR)
except Exception:
pass
def isClosed(self) -> bool:
return self._closed
def isConnected(self) -> bool:
return self._sock is not None
def getRemoteSocketAddress(self) -> Optional[LocalSocketAddress]:
return self._address
def toString(self) -> str:
return f"LocalSocket(address={self._address})"
# ═══════════════════════════════════════════════════════════════
# NetworkUtils
# ═══════════════════════════════════════════════════════════════
class NetworkUtils(Stub):
"""NetworkUtils — 网络工具"""
@staticmethod
def getNetworkPartOfAddress(addr: str, prefix_len: int) -> str:
return addr
@staticmethod
def intToInetAddress(host_address: int) -> str:
try:
return socket.inet_ntoa(struct.pack("!I", host_address & 0xFFFFFFFF))
except Exception:
return "0.0.0.0"
@staticmethod
def inetAddressToInt(addr: str) -> int:
try:
return struct.unpack("!I", socket.inet_aton(addr))[0]
except Exception:
return 0
@staticmethod
def isIPv6ULA(addr: str) -> bool:
return addr.startswith("fc") or addr.startswith("fd")
@staticmethod
def isIPv6GlobalMulticast(addr: str) -> bool:
return addr.startswith("ff")
@staticmethod
def prefixLengthToSubnetMask(prefix_len: int) -> int:
return (0xFFFFFFFF << (32 - prefix_len)) & 0xFFFFFFFF if prefix_len <= 32 else 0
# ═══════════════════════════════════════════════════════════════
# InetAddresses
# ═══════════════════════════════════════════════════════════════
class InetAddresses(Stub):
"""InetAddresses — 地址解析工具"""
@staticmethod
def isNumeric(address: str) -> bool:
try:
socket.inet_aton(address)
return True
except OSError:
return False
@staticmethod
def parseNumericAddress(address: str) -> str:
return address
@staticmethod
def getAddressFamily(address: str) -> int:
return 2 # AF_INET
# ═══════════════════════════════════════════════════════════════
# NetworkSpecifier / MatchAllNetworkSpecifier
# ═══════════════════════════════════════════════════════════════
class NetworkSpecifier(Stub):
"""NetworkSpecifier — 网络说明符基类"""
def __init__(self):
super().__init__()
def satisfiedBy(self, other: "NetworkSpecifier") -> bool:
return self is other
def equals(self, other) -> bool:
return self is other
def hashCode(self) -> int:
return hash(self.__class__.__name__) & 0x7FFFFFFF
class MatchAllNetworkSpecifier(NetworkSpecifier):
"""匹配所有网络的说明符"""
def __init__(self):
super().__init__()
def satisfiedBy(self, other: NetworkSpecifier) -> bool:
return True
def toString(self) -> str:
return "MatchAllNetworkSpecifier"
class WifiAwareNetworkSpecifier(NetworkSpecifier):
"""WiFi Aware 网络说明符"""
def __init__(self, role: int, client_id: int, peer_id: int, peer_mac: Optional[str] = None):
super().__init__()
self._role = role
self._client_id = client_id
self._peer_id = peer_id
self._peer_mac = peer_mac
def getRole(self) -> int:
return self._role
def getClientId(self) -> int:
return self._client_id
def toString(self) -> str:
return f"WifiAwareNetworkSpecifier(role={self._role}, client={self._client_id})"
class EthernetNetworkSpecifier(NetworkSpecifier):
"""以太网网络说明符"""
def __init__(self, data: Optional[str] = None):
super().__init__()
self._data = data
def getData(self) -> Optional[str]:
return self._data
def toString(self) -> str:
return f"EthernetNetworkSpecifier(data={self._data})"
class TelephonyNetworkSpecifier(NetworkSpecifier):
"""电话网络说明符"""
def __init__(self, subscription_id: int = 0):
super().__init__()
self._subscription_id = subscription_id
def getSubscriptionId(self) -> int:
return self._subscription_id
def toString(self) -> str:
return f"TelephonyNetworkSpecifier(subId={self._subscription_id})"
# ═══════════════════════════════════════════════════════════════
# NetworkState
# ═══════════════════════════════════════════════════════════════
class NetworkState(Stub):
"""NetworkState — 网络状态"""
def __init__(self, network: Optional["Network"] = None,
network_info: Optional["NetworkInfo"] = None,
link_properties: Optional["LinkProperties"] = None,
network_capabilities: Optional["NetworkCapabilities"] = None,
subscriber_id: Optional[str] = None):
super().__init__()
self.network = network
self.network_info = network_info
self.link_properties = link_properties
self.network_capabilities = network_capabilities
self.subscriber_id = subscriber_id
def getNetwork(self) -> Optional["Network"]:
return self.network
def getNetworkInfo(self) -> Optional["NetworkInfo"]:
return self.network_info
def getLinkProperties(self) -> Optional["LinkProperties"]:
return self.link_properties
def getNetworkCapabilities(self) -> Optional["NetworkCapabilities"]:
return self.network_capabilities
def toString(self) -> str:
return f"NetworkState(network={self.network})"
# ═══════════════════════════════════════════════════════════════
# NetworkInfo
# ═══════════════════════════════════════════════════════════════
class NetworkInfo(Stub):
"""NetworkInfo — 网络连接信息"""
class DetailedState(Stub):
CONNECTED = "CONNECTED"
CONNECTING = "CONNECTING"
DISCONNECTED = "DISCONNECTED"
SUSPENDED = "SUSPENDED"
IDLE = "IDLE"
FAILED = "FAILED"
BLOCKED = "BLOCKED"
VERIFYING_POOR_CONNECTION = "VERIFYING_POOR_CONNECTION"
CAPTIVE_PORTAL_CHECK = "CAPTIVE_PORTAL_CHECK"
OBTAINING_IPADDR = "OBTAINING_IPADDR"
class State(Stub):
CONNECTED = "CONNECTED"
CONNECTING = "CONNECTING"
DISCONNECTED = "DISCONNECTED"
SUSPENDED = "SUSPENDED"
UNKNOWN = "UNKNOWN"
def __init__(self, type_: int = 0, subtype: int = 0,
state: str = "DISCONNECTED", detailed_state: str = "DISCONNECTED",
type_name: str = "", subtype_name: str = "",
extra_info: Optional[str] = None, reason: Optional[str] = None,
failover: bool = False, roaming: bool = False,
is_available: bool = True):
super().__init__()
self._type = type_
self._subtype = subtype
self._state = state
self._detailed_state = detailed_state
self._type_name = type_name or self._default_type_name(type_)
self._subtype_name = subtype_name
self._extra_info = extra_info
self._reason = reason
self._failover = failover
self._roaming = roaming
self._is_available = is_available
self._is_connected = (state == "CONNECTED")
self._is_connecting = (state == "CONNECTING")
def _default_type_name(self, type_: int) -> str:
names = {
0: "MOBILE", 1: "WIFI", 6: "WIMAX", 7: "BLUETOOTH",
9: "ETHERNET", 17: "VPN",
}
return names.get(type_, "UNKNOWN")
def getType(self) -> int:
return self._type
def getSubtype(self) -> int:
return self._subtype
def isConnected(self) -> bool:
return self._state == "CONNECTED"
def isConnectedOrConnecting(self) -> bool:
return self._state in ("CONNECTED", "CONNECTING")
def getState(self) -> str:
return self._state
def getDetailedState(self) -> str:
return self._detailed_state
def getTypeName(self) -> str:
return self._type_name
def getSubtypeName(self) -> str:
return self._subtype_name
def getExtraInfo(self) -> Optional[str]:
return self._extra_info
def getReason(self) -> Optional[str]:
return self._reason
def getFailover(self) -> bool:
return self._failover
def getRoaming(self) -> bool:
return self._roaming
def isAvailable(self) -> bool:
return self._is_available
def isFailover(self) -> bool:
return self._failover
def isRoaming(self) -> bool:
return self._roaming
def toString(self) -> str:
return (f"NetworkInfo(type={self._type_name}[{self._type}], "
f"state={self._state}/{self._detailed_state}, "
f"roaming={self._roaming}, available={self._is_available})")
def hashCode(self) -> int:
return self._type
def equals(self, other) -> bool:
if isinstance(other, NetworkInfo):
return self._type == other._type and self._state == other._state
return False
# ═══════════════════════════════════════════════════════════════
# Network
# ═══════════════════════════════════════════════════════════════
class Network(Stub):
"""Network — 网络句柄"""
def __init__(self, net_id: int = 0):
super().__init__()
self._net_id = net_id
def getNetworkHandle(self) -> int:
return self._net_id
def getNetworkCapabilities(self) -> Optional["NetworkCapabilities"]:
return None
def getLinkProperties(self) -> Optional["LinkProperties"]:
return None
def bindSocket(self, socket_obj: Any):
pass
def bindSocketToNetwork(self, socket_obj: Any) -> bool:
return True
def equals(self, other) -> bool:
if isinstance(other, Network):
return self._net_id == other._net_id
return False
def hashCode(self) -> int:
return self._net_id
def toString(self) -> str:
return f"Network({self._net_id})"
# ═══════════════════════════════════════════════════════════════
# NetworkCapabilities
# ═══════════════════════════════════════════════════════════════
class NetworkCapabilities(Stub):
"""NetworkCapabilities — 网络能力"""
NET_CAPABILITY_INTERNET = 12
NET_CAPABILITY_NOT_METERED = 11
NET_CAPABILITY_NOT_ROAMING = 18
NET_CAPABILITY_VALIDATED = 16
NET_CAPABILITY_CAPTIVE_PORTAL = 17
NET_CAPABILITY_NOT_VPN = 15
NET_CAPABILITY_MMS = 0
NET_CAPABILITY_SUPL = 1
NET_CAPABILITY_DUN = 2
NET_CAPABILITY_FOTA = 3
NET_CAPABILITY_IMS = 4
NET_CAPABILITY_CBS = 5
NET_CAPABILITY_WIFI_P2P = 6
NET_CAPABILITY_IA = 7
NET_CAPABILITY_RCS = 8
NET_CAPABILITY_XCAP = 9
NET_CAPABILITY_EIMS = 10
NET_CAPABILITY_NOT_CONGESTED = 13
NET_CAPABILITY_NOT_RESTRICTED = 14
NET_CAPABILITY_TRUSTED = 19
NET_CAPABILITY_NOT_VCN_MANAGED = 20
NET_CAPABILITY_PRIORITIZE_LATENCY = 21
NET_CAPABILITY_PRIORITIZE_BANDWIDTH = 22
TRANSPORT_WIFI = 1
TRANSPORT_CELLULAR = 0
TRANSPORT_ETHERNET = 3
TRANSPORT_BLUETOOTH = 2
TRANSPORT_VPN = 4
TRANSPORT_WIFI_AWARE = 5
TRANSPORT_LOWPAN = 6
TRANSPORT_TEST = 7
SIGNAL_STRENGTH_UNSPECIFIED = -2147483648
def __init__(self):
super().__init__()
self._capabilities: Set[int] = set()
self._transports: Set[int] = set()
self._link_upstream_bw = 0
self._link_downstream_bw = 0
self._signal_strength = self.SIGNAL_STRENGTH_UNSPECIFIED
self._network_specifier: Optional[NetworkSpecifier] = None
self._owner_uid = -1
self._ssid: Optional[str] = None
self._administrator_uids: List[int] = []
self._enterprise_ids: List[int] = []
def hasCapability(self, capability: int) -> bool:
return capability in self._capabilities
def hasTransport(self, transport: int) -> bool:
return transport in self._transports
def addCapability(self, capability: int) -> "NetworkCapabilities":
self._capabilities.add(capability)
return self
def removeCapability(self, capability: int) -> "NetworkCapabilities":
self._capabilities.discard(capability)
return self
def getCapabilities(self) -> List[int]:
return sorted(self._capabilities)
def getLinkUpstreamBandwidthKbps(self) -> int:
return self._link_upstream_bw
def getLinkDownstreamBandwidthKbps(self) -> int:
return self._link_downstream_bw
def setLinkUpstreamBandwidthKbps(self, bw: int):
self._link_upstream_bw = bw
def setLinkDownstreamBandwidthKbps(self, bw: int):
self._link_downstream_bw = bw
def addTransportType(self, transport: int) -> "NetworkCapabilities":
self._transports.add(transport)
return self
def removeTransportType(self, transport: int) -> "NetworkCapabilities":
self._transports.discard(transport)
return self
def getTransportInfo(self) -> Optional[Any]:
return None
def setNetworkSpecifier(self, specifier: NetworkSpecifier):
self._network_specifier = specifier
def getNetworkSpecifier(self) -> Optional[NetworkSpecifier]:
return self._network_specifier
def getSignalStrength(self) -> int:
return self._signal_strength
def setSignalStrength(self, strength: int):
self._signal_strength = strength
def getOwnerUid(self) -> int:
return self._owner_uid
def setOwnerUid(self, uid: int):
self._owner_uid = uid
def getSsid(self) -> Optional[str]:
return self._ssid
def setSsid(self, ssid: str):
self._ssid = ssid
def getAdministratorUids(self) -> List[int]:
return list(self._administrator_uids)
def setAdministratorUids(self, uids: List[int]):
self._administrator_uids = list(uids)
def equals(self, other) -> bool:
if isinstance(other, NetworkCapabilities):
return (self._capabilities == other._capabilities
and self._transports == other._transports)
return False
def hashCode(self) -> int:
return hash(frozenset(self._capabilities)) ^ hash(frozenset(self._transports))
def toString(self) -> str:
caps = ",".join(str(c) for c in sorted(self._capabilities))
trans = ",".join(str(t) for t in sorted(self._transports))
return f"NetworkCapabilities(caps=[{caps}], transports=[{trans}])"
# ═══════════════════════════════════════════════════════════════
# NetworkRequest
# ═══════════════════════════════════════════════════════════════
class NetworkRequest(Stub):
"""NetworkRequest — 网络请求"""
class Builder(Stub):
"""Builder 模式"""
def __init__(self):
super().__init__()
self._capabilities: Set[int] = set()
self._transports: Set[int] = set()
self._network_specifier: Optional[NetworkSpecifier] = None
self._clear_capabilities: Set[int] = set()
def addCapability(self, capability: int) -> "NetworkRequest.Builder":
self._capabilities.add(capability)
return self
def removeCapability(self, capability: int) -> "NetworkRequest.Builder":
self._capabilities.discard(capability)
self._clear_capabilities.add(capability)
return self
def addTransportType(self, transport: int) -> "NetworkRequest.Builder":
self._transports.add(transport)
return self
def removeTransportType(self, transport: int) -> "NetworkRequest.Builder":
self._transports.discard(transport)
return self
def setNetworkSpecifier(self, specifier: NetworkSpecifier) -> "NetworkRequest.Builder":
self._network_specifier = specifier
return self
def build(self) -> "NetworkRequest":
request = NetworkRequest()
request._capabilities = set(self._capabilities)
request._transports = set(self._transports)
request._network_specifier = self._network_specifier
request._request_id = id(request)
return request
def __init__(self):
super().__init__()
self._capabilities: Set[int] = set()
self._transports: Set[int] = set()
self._network_specifier: Optional[NetworkSpecifier] = None
self._request_id = 0
def hasCapability(self, capability: int) -> bool:
return capability in self._capabilities
def hasTransport(self, transport: int) -> bool:
return transport in self._transports
def getCapabilities(self) -> Set[int]:
return set(self._capabilities)
def getTransports(self) -> Set[int]:
return set(self._transports)
def getNetworkSpecifier(self) -> Optional[NetworkSpecifier]:
return self._network_specifier
def getRequestId(self) -> int:
return self._request_id
def toString(self) -> str:
return f"NetworkRequest(id={self._request_id})"
# ═══════════════════════════════════════════════════════════════
# NetworkCallback
# ═══════════════════════════════════════════════════════════════
class NetworkCallback(Stub):
"""NetworkCallback — 网络回调"""
def __init__(self, flags: int = 0):
super().__init__()
self._flags = flags
def onAvailable(self, network: Network):
pass
def onLost(self, network: Network):
pass
def onCapabilitiesChanged(self, network: Network, capabilities: NetworkCapabilities):
pass
def onLinkPropertiesChanged(self, network: Network, link_properties: "LinkProperties"):
pass
def onNetworkSuspended(self, network: Network):
pass
def onResumed(self, network: Network):
pass
def onBlockedStatusChanged(self, network: Network, blocked: bool):
pass
def onLosing(self, network: Network, max_ms_to_live: int):
pass
def onPreCheck(self, network: Network) -> bool:
return True
def onUnavailable(self):
pass
# ═══════════════════════════════════════════════════════════════
# LinkAddress
# ═══════════════════════════════════════════════════════════════
class LinkAddress(Stub):
"""LinkAddress — 链接地址"""
def __init__(self, address: str = "0.0.0.0",
prefix_length: int = 24,
flags: int = 0,
scope: int = 0):
super().__init__()
self._address = address
self._prefix_length = prefix_length
self._flags = flags
self._scope = scope
self._is_ipv6 = ":" in address
def getAddress(self) -> str:
return self._address
def getPrefixLength(self) -> int:
return self._prefix_length
def getFlags(self) -> int:
return self._flags
def getScope(self) -> int:
return self._scope
def isIpv6(self) -> bool:
return self._is_ipv6
def toString(self) -> str:
return f"{self._address}/{self._prefix_length}"
def equals(self, other) -> bool:
if isinstance(other, LinkAddress):
return self._address == other._address
return False
def hashCode(self) -> int:
return hash(self._address) & 0x7FFFFFFF
# ═══════════════════════════════════════════════════════════════
# RouteInfo
# ═══════════════════════════════════════════════════════════════
class RouteInfo(Stub):
"""RouteInfo — 路由信息"""
RTN_UNICAST = 1
RTN_UNREACHABLE = 2
RTN_BROADCAST = 3
RTN_THROW = 4
RTN_OTHER = 5
def __init__(self, destination: Optional[LinkAddress] = None,
gateway: str = "",
interface_name: str = "",
is_default: bool = False,
type_: int = RTN_UNICAST):
super().__init__()
self._destination = destination
self._gateway = gateway
self._interface_name = interface_name
self._is_default = is_default
self._type = type_
def getDestination(self) -> Optional[LinkAddress]:
return self._destination
def getGateway(self) -> str:
return self._gateway
def getInterface(self) -> str:
return self._interface_name
def getType(self) -> int:
return self._type
def isDefaultRoute(self) -> bool:
return self._is_default
def isIPv4Default(self) -> bool:
return self._is_default and self._destination is not None and not self._destination.isIpv6()
def isIPv6Default(self) -> bool:
return self._is_default and self._destination is not None and self._destination.isIpv6()
def toString(self) -> str:
dest = str(self._destination) if self._destination else "0.0.0.0/0"
gw = self._gateway or "0.0.0.0"
return f"RouteInfo({dest} -> {gw} via {self._interface_name})"
def equals(self, other) -> bool:
if isinstance(other, RouteInfo):
return (self._gateway == other._gateway
and self._interface_name == other._interface_name)
return False
def hashCode(self) -> int:
return hash(self._gateway) & 0x7FFFFFFF
# ═══════════════════════════════════════════════════════════════
# LinkProperties
# ═══════════════════════════════════════════════════════════════
class LinkProperties(Stub):
"""LinkProperties — 链接属性"""
def __init__(self):
super().__init__()
self._link_addresses: List[LinkAddress] = []
self._routes: List[RouteInfo] = []
self._dns_servers: List[str] = []
self._interface_name: str = ""
self._mtu: int = 1500
self._http_proxy: Optional[ProxyInfo] = None
self._private_dns_server_name: str = ""
self._private_dns_active: bool = False
self._domains: str = ""
self._pcscf: List[str] = []
def getLinkAddresses(self) -> List[LinkAddress]:
return list(self._link_addresses)
def getRoutes(self) -> List[RouteInfo]:
return list(self._routes)
def getDnsServers(self) -> List[str]:
return list(self._dns_servers)
def getInterfaceName(self) -> str:
return self._interface_name
def setInterfaceName(self, name: str):
self._interface_name = name
def addLinkAddress(self, address: LinkAddress):
if address not in self._link_addresses:
self._link_addresses.append(address)
def removeLinkAddress(self, address: LinkAddress):
self._link_addresses = [a for a in self._link_addresses if a != address]
def addRoute(self, route: RouteInfo):
if route not in self._routes:
self._routes.append(route)
def removeRoute(self, route: RouteInfo):
self._routes = [r for r in self._routes if r != route]
def addDnsServer(self, dns: str):
if dns not in self._dns_servers:
self._dns_servers.append(dns)
def removeDnsServer(self, dns: str):
self._dns_servers = [d for d in self._dns_servers if d != dns]
def getMtu(self) -> int:
return self._mtu
def setMtu(self, mtu: int):
self._mtu = mtu
def getHttpProxy(self) -> Optional["ProxyInfo"]:
return self._http_proxy
def setHttpProxy(self, proxy: "ProxyInfo"):
self._http_proxy = proxy
def getPrivateDnsServerName(self) -> str:
return self._private_dns_server_name
def setPrivateDnsServerName(self, name: str):
self._private_dns_server_name = name
self._private_dns_active = bool(name)
def isPrivateDnsActive(self) -> bool:
return self._private_dns_active
def setLinkAddresses(self, addresses: List[LinkAddress]):
self._link_addresses = list(addresses)
def setDnsServers(self, dns_servers: List[str]):
self._dns_servers = list(dns_servers)
def getDomains(self) -> str:
return self._domains
def setDomains(self, domains: str):
self._domains = domains
def getPcscf(self) -> List[str]:
return list(self._pcscf)
def addPcscf(self, addr: str):
if addr not in self._pcscf:
self._pcscf.append(addr)
def setMtuV4(int_mtu: int):
pass
def setMtuV6(int_mtu: int):
pass
def clearLinkAddresses(self):
self._link_addresses.clear()
def clearRoutes(self):
self._routes.clear()
def clearDnsServers(self):
self._dns_servers.clear()
def toString(self) -> str:
iface = self._interface_name or "none"
return f"LinkProperties(iface={iface}, mtu={self._mtu}, dns={self._dns_servers})"
def equals(self, other) -> bool:
if isinstance(other, LinkProperties):
return (self._interface_name == other._interface_name
and self._mtu == other._mtu)
return False
def hashCode(self) -> int:
return hash(self._interface_name) & 0x7FFFFFFF
# ═══════════════════════════════════════════════════════════════
# DnsResolver
# ═══════════════════════════════════════════════════════════════
class DnsResolver(Stub):
"""DnsResolver — DNS 解析器"""
CLASS_IN = 1
TYPE_A = 1
TYPE_AAAA = 28
@staticmethod
def getInstance() -> "DnsResolver":
return DnsResolver()
def query(self, hostname: str, callback: Any, flags: int = 0):
try:
result = socket.getaddrinfo(hostname, None)
callback.onAnswer(None, 0, result)
except socket.gaierror:
callback.onAnswer(None, -1, None)
# ═══════════════════════════════════════════════════════════════
# DhcpInfo
# ═══════════════════════════════════════════════════════════════
class DhcpInfo(Stub):
"""DhcpInfo — DHCP 信息"""
def __init__(self):
super().__init__()
self.ipAddress = 0
self.gateway = 0
self.netmask = 0
self.dns1 = 0
self.dns2 = 0
self.serverAddress = 0
self.leaseDuration = 0
def toString(self) -> str:
return (f"DhcpInfo(ip={self.ipAddress}, gw={self.gateway}, "
f"mask={self.netmask}, dns1={self.dns1}, dns2={self.dns2})")
# ═══════════════════════════════════════════════════════════════
# SSLSessionCache
# ═══════════════════════════════════════════════════════════════
class SSLSessionCache(Stub):
"""SSLSessionCache — SSL 会话缓存"""
def __init__(self, context_or_dir: Any):
super().__init__()
self._dir: Optional[str] = None
if isinstance(context_or_dir, str):
self._dir = context_or_dir
else:
self._dir = os.path.join(_app_data_dir, "ssl_cache")
def getDir(self) -> Optional[str]:
return self._dir
# ═══════════════════════════════════════════════════════════════
# TrafficStats
# ═══════════════════════════════════════════════════════════════
class TrafficStats(Stub):
"""TrafficStats — 流量统计"""
UID_TETHERING = "UNSUPPORTED"
UNSUPPORTED = -1
TAG_SYSTEM_UPDATE = 0
@staticmethod
def getMobileTxBytes() -> int:
return 0
@staticmethod
def getMobileRxBytes() -> int:
return 0
@staticmethod
def getTotalTxBytes() -> int:
return 0
@staticmethod
def getTotalRxBytes() -> int:
return 0
@staticmethod
def getMobileTxPackets() -> int:
return 0
@staticmethod
def getMobileRxPackets() -> int:
return 0
@staticmethod
def getTotalTxPackets() -> int:
return 0
@staticmethod
def getTotalRxPackets() -> int:
return 0
@staticmethod
def getUidTxBytes(uid: int) -> int:
return 0
@staticmethod
def getUidRxBytes(uid: int) -> int:
return 0
@staticmethod
def getUidTxPackets(uid: int) -> int:
return 0
@staticmethod
def getUidRxPackets(uid: int) -> int:
return 0
@staticmethod
def getUidTcpTxBytes(uid: int) -> int:
return 0
@staticmethod
def getUidTcpRxBytes(uid: int) -> int:
return 0
@staticmethod
def getUidUdpTxBytes(uid: int) -> int:
return 0
@staticmethod
def getUidUdpRxBytes(uid: int) -> int:
return 0
@staticmethod
def setThreadStatsTag(tag: int):
pass
@staticmethod
def clearThreadStatsTag():
pass
@staticmethod
def getThreadStatsTag() -> int:
return 0
@staticmethod
def setThreadStatsUid(uid: int):
pass
@staticmethod
def clearThreadStatsUid():
pass
@staticmethod
def incrementOperationCount(uid: int, tag: int, count: int):
pass
# ═══════════════════════════════════════════════════════════════
# Proxy / ProxyInfo
# ═══════════════════════════════════════════════════════════════
class Proxy(Stub):
"""Proxy — 系统代理信息"""
@staticmethod
def getDefaultHost() -> Optional[str]:
return None
@staticmethod
def getDefaultPort() -> int:
return -1
class ProxyInfo(Stub):
"""ProxyInfo — 代理详细信息"""
def __init__(self, host: str = "", port: int = 0,
pac_file_url: Optional[str] = None,
username: str = "", password: str = "",
exclusion_list: Optional[List[str]] = None):
super().__init__()
self._host = host
self._port = port
self._pac_file_url = pac_file_url
self._username = username
self._password = password
self._exclusion_list = list(exclusion_list) if exclusion_list else []
self._valid = bool(host) or bool(pac_file_url)
def getHost(self) -> str:
return self._host
def getPort(self) -> int:
return self._port
def getPacFileUrl(self) -> Optional[str]:
return self._pac_file_url
def getUsername(self) -> str:
return self._username
def getPassword(self) -> str:
return self._password
def getExclusionList(self) -> List[str]:
return list(self._exclusion_list)
def isValid(self) -> bool:
return self._valid
def makeProxy(self, url: str):
return self
def equals(self, other) -> bool:
if isinstance(other, ProxyInfo):
return self._host == other._host and self._port == other._port
return False
def hashCode(self) -> int:
return hash(self._host) ^ self._port
def toString(self) -> str:
pac = f" pac={self._pac_file_url}" if self._pac_file_url else ""
return f"ProxyInfo(host={self._host}:{self._port}{pac})"
# ═══════════════════════════════════════════════════════════════
# ConnectivityManager
# ═══════════════════════════════════════════════════════════════
class ConnectivityManager(Stub):
"""ConnectivityManager — 连接管理器"""
TYPE_WIFI = 1
TYPE_MOBILE = 0
TYPE_BLUETOOTH = 7
TYPE_ETHERNET = 9
TYPE_VPN = 17
TYPE_WIMAX = 6
TYPE_DUMMY = 8
TYPE_PROXY = 16
CONNECTIVITY_ACTION = "android.net.conn.CONNECTIVITY_CHANGE"
DEFAULT_NETWORK_PREFERENCE = 1
def __init__(self):
super().__init__()
self._networks: Dict[int, "Network"] = {}
self._callbacks: List[Tuple[NetworkCallback, Optional[NetworkRequest]]] = []
self._active_network_info = NetworkInfo(
type_=self.TYPE_WIFI, state="CONNECTED", detailed_state="CONNECTED",
type_name="WIFI", is_available=True,
)
self._net_id_counter = 0
self._network_caps: Dict[int, NetworkCapabilities] = {}
self._link_props: Dict[int, LinkProperties] = {}
def getActiveNetworkInfo(self) -> Optional[NetworkInfo]:
return self._active_network_info
def getNetworkInfo(self, network_type: int) -> Optional[NetworkInfo]:
network = self._networks.get(network_type)
if network:
return NetworkInfo(type_=network_type, state="CONNECTED", is_available=True)
return self._active_network_info
def getNetworkInfoForUid(self, network: Network, uid: int) -> Optional[NetworkInfo]:
return self._active_network_info
def getAllNetworks(self) -> List[Network]:
return list(self._networks.values())
def getActiveNetwork(self) -> Optional[Network]:
if self._networks:
nid = next(iter(self._networks))
return self._networks[nid]
net = Network(1)
self._networks[1] = net
return net
def getActiveNetworkForUid(self, uid: int) -> Optional[Network]:
return self.getActiveNetwork()
def getNetworkCapabilities(self, network: Network) -> Optional[NetworkCapabilities]:
caps = self._network_caps.get(network.getNetworkHandle())
if caps is None:
caps = NetworkCapabilities()
caps.addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
caps.addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_RESTRICTED)
caps.addCapability(NetworkCapabilities.NET_CAPABILITY_TRUSTED)
caps.addTransportType(NetworkCapabilities.TRANSPORT_WIFI)
self._network_caps[network.getNetworkHandle()] = caps
return caps
def getLinkProperties(self, network: Network) -> Optional[LinkProperties]:
lp = self._link_props.get(network.getNetworkHandle())
if lp is None:
lp = LinkProperties()
self._link_props[network.getNetworkHandle()] = lp
return lp
def requestNetwork(self, request: NetworkRequest, callback: NetworkCallback, timeout_ms: int = 0,
handler: Any = None, network_specifier: int = 0) -> None:
self._callbacks.append((callback, request))
net = self.getActiveNetwork()
if net:
try:
callback.onAvailable(net)
except Exception:
pass
def registerDefaultNetworkCallback(self, callback: NetworkCallback, handler: Any = None) -> None:
self._callbacks.append((callback, None))
net = self.getActiveNetwork()
if net:
try:
callback.onAvailable(net)
except Exception:
pass
def registerNetworkCallback(self, request: NetworkRequest, callback: NetworkCallback,
handler: Any = None) -> None:
self._callbacks.append((callback, request))
net = self.getActiveNetwork()
if net:
try:
callback.onAvailable(net)
except Exception:
pass
def unregisterNetworkCallback(self, callback_or_request: Any):
if isinstance(callback_or_request, NetworkCallback):
self._callbacks = [
(cb, req) for cb, req in self._callbacks
if cb is not callback_or_request
]
elif isinstance(callback_or_request, NetworkRequest):
self._callbacks = [
(cb, req) for cb, req in self._callbacks
if req is not callback_or_request
]
def bindProcessToNetwork(self, network: Optional[Network]) -> bool:
return True
def getProcessDefaultNetwork(self) -> Optional[Network]:
return self.getActiveNetwork()
@staticmethod
def setProcessDefaultNetwork(network: Optional[Network]) -> bool:
return True
def getAllNetworkInfo(self) -> List[NetworkInfo]:
return [self._active_network_info]
def getNetworkInfoForProcess(self) -> Optional[NetworkInfo]:
return self._active_network_info
def isDefaultNetworkActive(self) -> bool:
return True
def addDefaultNetworkActiveListener(self, listener: Any):
pass
def removeDefaultNetworkActiveListener(self, listener: Any):
pass
def reportBadNetwork(self, network: Network):
pass
def setNetworkPreference(self, preference: int):
pass
def getNetworkPreference(self) -> int:
return self.DEFAULT_NETWORK_PREFERENCE
def startCaptivePortalApp(self, network: Network):
pass
def setProcessDefaultNetworkForUid(self, uid: int) -> bool:
return True
def registerNetworkAgent(self, agent: Any) -> int:
return 0
def toString(self) -> str:
return f"ConnectivityManager(networks={len(self._networks)})"
# ═══════════════════════════════════════════════════════════════
# __all__
# ═══════════════════════════════════════════════════════════════
__all__ = [
"Uri",
"Uri_Builder",
"ConnectivityManager",
"NetworkInfo",
"Network",
"NetworkCapabilities",
"NetworkRequest",
"NetworkCallback",
"LinkProperties",
"LinkAddress",
"RouteInfo",
"DnsResolver",
"DhcpInfo",
"InetAddresses",
"SSLSessionCache",
"TrafficStats",
"Proxy",
"ProxyInfo",
"NetworkSpecifier",
"MatchAllNetworkSpecifier",
"WifiAwareNetworkSpecifier",
"EthernetNetworkSpecifier",
"TelephonyNetworkSpecifier",
"NetworkState",
"LocalSocket",
"LocalSocketAddress",
"NetworkUtils",
]