shuiyu1123
a51854455b
- android/ 167个模块 (app/content/os/view/widget/graphics/media/hardware/net/telephony/provider等) - java/ 35个模块 (lang/io/util/net/nio/security/crypto/sql/beans/rmi/awt等) - javax/ 4个模块 (net/crypto/security/sql) - org/ 5个模块 (json/w3c_dom/xml_sax/xml_pull/apache_http) - dalvik/ 1个模块 (system) - ~3053个类/接口，方法签名完全对齐Android SDK - 桌面跨平台运行 (Windows/macOS/Linux) - 修复所有导入错误和Python关键字冲突
406 lines
14 KiB
Python
406 lines
14 KiB
Python
"""
|
|
Android net.ssl 组件 — SSLSessionCache, SSLEngine, SSLSession, SSLContext, SSLParameters 等
|
|
方法签名完全对齐 Android SDK
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import ssl
|
|
import threading
|
|
import time as _time
|
|
from typing import Any, Dict, List, Optional, Set, Tuple
|
|
|
|
from .._core import Stub, AndroidOnlyError
|
|
from ..java.lang import RuntimeException
|
|
|
|
|
|
IllegalArgumentException = type("IllegalArgumentException", (RuntimeException,), {})
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# SSLSessionCache
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class SSLSessionCache(Stub):
|
|
"""SSLSessionCache — SSL 会话缓存"""
|
|
|
|
def __init__(self, path: Optional[str] = None):
|
|
super().__init__()
|
|
self._path = path
|
|
self._cache: Dict[str, Any] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def install(self, ssl_context: "SSLContext"):
|
|
pass
|
|
|
|
@staticmethod
|
|
def get(ssl_session: Any) -> Optional["SSLSessionCache"]:
|
|
return None
|
|
|
|
def toString(self) -> str:
|
|
return f"SSLSessionCache@{hex(id(self))}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# SSLSession
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class SSLSession(Stub):
|
|
"""SSLSession — SSL 会话"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._creation_time: int = int(_time.time() * 1000)
|
|
self._last_accessed_time: int = self._creation_time
|
|
self._application_buffer_size: int = 16384
|
|
self._packet_buffer_size: int = 16640
|
|
self._cipher_suite: str = "TLS_AES_128_GCM_SHA256"
|
|
self._protocol: str = "TLSv1.3"
|
|
self._id: bytes = os.urandom(32)
|
|
self._peer_host: str = ""
|
|
self._peer_port: int = 443
|
|
self._local_certificates: Optional[List[Any]] = None
|
|
self._peer_certificates: Optional[List[Any]] = None
|
|
self._local_principal: Any = None
|
|
self._peer_principal: Any = None
|
|
self._values: Dict[str, Any] = {}
|
|
self._session_context: Optional["SSLContext"] = None
|
|
|
|
def getApplicationBufferSize(self) -> int:
|
|
return self._application_buffer_size
|
|
|
|
def getCipherSuite(self) -> str:
|
|
return self._cipher_suite
|
|
|
|
def getCreationTime(self) -> int:
|
|
return self._creation_time
|
|
|
|
def getId(self) -> bytes:
|
|
return self._id
|
|
|
|
def getLastAccessedTime(self) -> int:
|
|
return self._last_accessed_time
|
|
|
|
def getLocalCertificates(self) -> Optional[List[Any]]:
|
|
return self._local_certificates
|
|
|
|
def getLocalPrincipal(self) -> Any:
|
|
return self._local_principal
|
|
|
|
def getPacketBufferSize(self) -> int:
|
|
return self._packet_buffer_size
|
|
|
|
def getPeerCertificates(self) -> Optional[List[Any]]:
|
|
return self._peer_certificates
|
|
|
|
def getPeerHost(self) -> str:
|
|
return self._peer_host
|
|
|
|
def getPeerPort(self) -> int:
|
|
return self._peer_port
|
|
|
|
def getPeerPrincipal(self) -> Any:
|
|
return self._peer_principal
|
|
|
|
def getProtocol(self) -> str:
|
|
return self._protocol
|
|
|
|
def getSessionContext(self) -> Optional["SSLContext"]:
|
|
return self._session_context
|
|
|
|
def getValue(self, name: str) -> Any:
|
|
return self._values.get(name)
|
|
|
|
def getValueNames(self) -> List[str]:
|
|
return list(self._values.keys())
|
|
|
|
def invalidate(self):
|
|
self._last_accessed_time = 0
|
|
self._values.clear()
|
|
|
|
def putValue(self, name: str, value: Any):
|
|
self._values[name] = value
|
|
|
|
def removeValue(self, name: str):
|
|
self._values.pop(name, None)
|
|
|
|
def toString(self) -> str:
|
|
return f"SSLSession{{protocol={self._protocol}, cipher={self._cipher_suite}}}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# SSLParameters
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class SSLParameters(Stub):
|
|
"""SSLParameters — SSL 参数"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._cipher_suites: List[str] = []
|
|
self._protocols: List[str] = []
|
|
self._want_client_auth: bool = False
|
|
self._need_client_auth: bool = False
|
|
self._algorithm_constraints: Any = None
|
|
self._endpoint_identification_algorithm: Optional[str] = None
|
|
self._server_names: List[Any] = []
|
|
self._sni_matchers: List[Any] = []
|
|
self._named_groups: List[str] = []
|
|
|
|
def getCipherSuites(self) -> List[str]:
|
|
return list(self._cipher_suites)
|
|
|
|
def setCipherSuites(self, cipherSuites: List[str]):
|
|
self._cipher_suites = list(cipherSuites)
|
|
|
|
def getProtocols(self) -> List[str]:
|
|
return list(self._protocols)
|
|
|
|
def setProtocols(self, protocols: List[str]):
|
|
self._protocols = list(protocols)
|
|
|
|
def getWantClientAuth(self) -> bool:
|
|
return self._want_client_auth
|
|
|
|
def setWantClientAuth(self, wantClientAuth: bool):
|
|
self._want_client_auth = wantClientAuth
|
|
|
|
def getNeedClientAuth(self) -> bool:
|
|
return self._need_client_auth
|
|
|
|
def setNeedClientAuth(self, needClientAuth: bool):
|
|
self._need_client_auth = needClientAuth
|
|
|
|
def getAlgorithmConstraints(self) -> Any:
|
|
return self._algorithm_constraints
|
|
|
|
def setAlgorithmConstraints(self, constraints: Any):
|
|
self._algorithm_constraints = constraints
|
|
|
|
def getEndpointIdentificationAlgorithm(self) -> Optional[str]:
|
|
return self._endpoint_identification_algorithm
|
|
|
|
def setEndpointIdentificationAlgorithm(self, algorithm: Optional[str]):
|
|
self._endpoint_identification_algorithm = algorithm
|
|
|
|
def getServerNames(self) -> List[Any]:
|
|
return list(self._server_names)
|
|
|
|
def setServerNames(self, serverNames: List[Any]):
|
|
self._server_names = list(serverNames)
|
|
|
|
def getSNIMatchers(self) -> List[Any]:
|
|
return list(self._sni_matchers)
|
|
|
|
def setSNIMatchers(self, sniMatchers: List[Any]):
|
|
self._sni_matchers = list(sniMatchers)
|
|
|
|
def getNamedGroups(self) -> List[str]:
|
|
return list(self._named_groups)
|
|
|
|
def setNamedGroups(self, namedGroups: List[str]):
|
|
self._named_groups = list(namedGroups)
|
|
|
|
def toString(self) -> str:
|
|
return f"SSLParameters@{hex(id(self))}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# SSLContext
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class SSLContext(Stub):
|
|
"""SSLContext — SSL 上下文"""
|
|
|
|
def __init__(self, protocol: str = "TLS"):
|
|
super().__init__()
|
|
self._protocol = protocol
|
|
self._ssl_context = ssl.create_default_context()
|
|
self._socket_factory: Any = None
|
|
self._server_socket_factory: Any = None
|
|
self._default_ssl_params: Optional[SSLParameters] = None
|
|
self._supported_ssl_params: Optional[SSLParameters] = None
|
|
|
|
@staticmethod
|
|
def getInstance(protocol: str = "TLS") -> "SSLContext":
|
|
ctx = SSLContext(protocol)
|
|
ctx._default_ssl_params = SSLParameters()
|
|
ctx._supported_ssl_params = SSLParameters()
|
|
return ctx
|
|
|
|
def init(self, keyManager: Any, trustManager: Any, random: Any):
|
|
pass
|
|
|
|
def getSocketFactory(self) -> Any:
|
|
if not self._socket_factory:
|
|
self._socket_factory = self._ssl_context.wrap_socket
|
|
return self._socket_factory
|
|
|
|
def getServerSocketFactory(self) -> Any:
|
|
if not self._server_socket_factory:
|
|
self._server_socket_factory = self._ssl_context.wrap_socket
|
|
return self._server_socket_factory
|
|
|
|
def createSSLEngine(self) -> "SSLEngine":
|
|
return SSLEngine(self)
|
|
|
|
def createSSLEngine(self, peerHost: str, peerPort: int) -> "SSLEngine":
|
|
return SSLEngine(self, peerHost, peerPort)
|
|
|
|
def getProtocol(self) -> str:
|
|
return self._protocol
|
|
|
|
def getSupportedSSLParameters(self) -> SSLParameters:
|
|
if not self._supported_ssl_params:
|
|
self._supported_ssl_params = SSLParameters()
|
|
return self._supported_ssl_params
|
|
|
|
def getDefaultSSLParameters(self) -> SSLParameters:
|
|
if not self._default_ssl_params:
|
|
self._default_ssl_params = SSLParameters()
|
|
return self._default_ssl_params
|
|
|
|
def toString(self) -> str:
|
|
return f"SSLContext{{protocol={self._protocol}}}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# SSLEngine
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class SSLEngine(Stub):
|
|
"""SSLEngine — SSL 引擎"""
|
|
|
|
HANDSHAKE_STATUS_NOT_STARTED = 0
|
|
HANDSHAKE_STATUS_IN_PROGRESS = 1
|
|
HANDSHAKE_STATUS_NEED_WRAP = 2
|
|
HANDSHAKE_STATUS_NEED_UNWRAP = 3
|
|
HANDSHAKE_STATUS_FINISHED = 4
|
|
HANDSHAKE_STATUS_NEED_TASK = 5
|
|
|
|
def __init__(self, context: SSLContext, peerHost: str = "", peerPort: int = 443):
|
|
super().__init__()
|
|
self._context = context
|
|
self._peer_host = peerHost
|
|
self._peer_port = peerPort
|
|
self._handshake_status = self.HANDSHAKE_STATUS_NOT_STARTED
|
|
self._use_client_mode: bool = True
|
|
self._need_client_auth: bool = False
|
|
self._want_client_auth: bool = False
|
|
self._enabled_cipher_suites: List[str] = [
|
|
"TLS_AES_128_GCM_SHA256", "TLS_AES_256_GCM_SHA384",
|
|
"TLS_CHACHA20_POLY1305_SHA256",
|
|
]
|
|
self._enabled_protocols: List[str] = ["TLSv1.3", "TLSv1.2"]
|
|
self._session = SSLSession()
|
|
self._delegated_tasks: List[Any] = []
|
|
self._application_protocol: Optional[str] = None
|
|
self._handshake_application_protocol: Optional[str] = None
|
|
self._hs_protocol_selector: Any = None
|
|
|
|
def beginHandshake(self):
|
|
self._handshake_status = self.HANDSHAKE_STATUS_IN_PROGRESS
|
|
|
|
def closeInbound(self):
|
|
pass
|
|
|
|
def closeOutbound(self):
|
|
pass
|
|
|
|
def getDelegatedTask(self) -> Optional[Any]:
|
|
if self._delegated_tasks:
|
|
return self._delegated_tasks.pop(0)
|
|
return None
|
|
|
|
def getEnabledCipherSuites(self) -> List[str]:
|
|
return list(self._enabled_cipher_suites)
|
|
|
|
def getEnabledProtocols(self) -> List[str]:
|
|
return list(self._enabled_protocols)
|
|
|
|
def getHandshakeStatus(self) -> int:
|
|
return self._handshake_status
|
|
|
|
def getNeedClientAuth(self) -> bool:
|
|
return self._need_client_auth
|
|
|
|
def getSession(self) -> SSLSession:
|
|
return self._session
|
|
|
|
def getSupportedCipherSuites(self) -> List[str]:
|
|
return list(self._enabled_cipher_suites)
|
|
|
|
def getSupportedProtocols(self) -> List[str]:
|
|
return list(self._enabled_protocols)
|
|
|
|
def getUseClientMode(self) -> bool:
|
|
return self._use_client_mode
|
|
|
|
def getWantClientAuth(self) -> bool:
|
|
return self._want_client_auth
|
|
|
|
def setEnabledCipherSuites(self, suites: List[str]):
|
|
self._enabled_cipher_suites = list(suites)
|
|
|
|
def setEnabledProtocols(self, protocols: List[str]):
|
|
self._enabled_protocols = list(protocols)
|
|
|
|
def setNeedClientAuth(self, needClientAuth: bool):
|
|
self._need_client_auth = needClientAuth
|
|
|
|
def setUseClientMode(self, useClientMode: bool):
|
|
self._use_client_mode = useClientMode
|
|
|
|
def setWantClientAuth(self, wantClientAuth: bool):
|
|
self._want_client_auth = wantClientAuth
|
|
|
|
def setSSLParameters(self, params: SSLParameters):
|
|
if params.getCipherSuites():
|
|
self._enabled_cipher_suites = params.getCipherSuites()
|
|
if params.getProtocols():
|
|
self._enabled_protocols = params.getProtocols()
|
|
self._need_client_auth = params.getNeedClientAuth()
|
|
self._want_client_auth = params.getWantClientAuth()
|
|
|
|
def getSSLParameters(self) -> SSLParameters:
|
|
params = SSLParameters()
|
|
params.setCipherSuites(self._enabled_cipher_suites)
|
|
params.setProtocols(self._enabled_protocols)
|
|
params.setNeedClientAuth(self._need_client_auth)
|
|
params.setWantClientAuth(self._want_client_auth)
|
|
return params
|
|
|
|
def wrap(self, src: Any, dst: Any) -> Any:
|
|
return src
|
|
|
|
def unwrap(self, src: Any, dst: Any) -> Any:
|
|
return src
|
|
|
|
def getApplicationProtocol(self) -> Optional[str]:
|
|
return self._application_protocol
|
|
|
|
def getHandshakeApplicationProtocol(self) -> Optional[str]:
|
|
return self._handshake_application_protocol
|
|
|
|
def setHandshakeApplicationProtocolSelector(self, selector: Any):
|
|
self._hs_protocol_selector = selector
|
|
|
|
def getHandshakeApplicationProtocolSelector(self) -> Any:
|
|
return self._hs_protocol_selector
|
|
|
|
def toString(self) -> str:
|
|
return f"SSLEngine{{handshake={self._handshake_status}, client={self._use_client_mode}}}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# __all__
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
__all__ = [
|
|
"SSLSessionCache",
|
|
"SSLEngine",
|
|
"SSLSession",
|
|
"SSLContext",
|
|
"SSLParameters",
|
|
] |