FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Consolidate SSL ciphers and use them in SslTransport by rytilahti · Pull Request #1798 · python-kasa/python-kasa · GitHub

Repository navigation

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (7) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
36 changes: 4 additions & 32 deletions kasa/transports/klaptransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@
import hashlib
import logging
import secrets
import ssl
import struct
import time
from asyncio import Future
Expand All @@ -67,6 +66,7 @@
from kasa.protocols.protocol import md5

from .basetransport import BaseTransport
from .sslcontext import get_ssl_context

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -97,17 +97,6 @@ class KlapTransport(BaseTransport):

SESSION_COOKIE_NAME = "TP_SESSIONID"
TIMEOUT_COOKIE_NAME = "TIMEOUT"
# Copy & paste from sslaestransport
CIPHERS = ":".join(
[
"AES256-GCM-SHA384",
"AES256-SHA256",
"AES128-GCM-SHA256",
"AES128-SHA256",
"AES256-SHA",
]
)
_ssl_context: ssl.SSLContext | None = None

def __init__(
self,
Expand Down Expand Up @@ -193,7 +182,7 @@ async def perform_handshake1(self) -> tuple[bytes, bytes, bytes]:
url = self._app_url / "handshake1"

response_status, response_data = await self._http_client.post(
url, data=payload, ssl=await self._get_ssl_context()
url, data=payload, ssl=await get_ssl_context()
)

if _LOGGER.isEnabledFor(logging.DEBUG):
Expand Down Expand Up @@ -305,7 +294,7 @@ async def perform_handshake2(
url,
data=payload,
cookies_dict=self._session_cookie,
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)

if _LOGGER.isEnabledFor(logging.DEBUG):
Expand Down Expand Up @@ -380,7 +369,7 @@ async def send(self, request: str) -> Generator[Future, None, dict[str, str]]:
params={"seq": seq},
data=payload,
cookies_dict=self._session_cookie,
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)

msg = (
Expand Down Expand Up @@ -457,23 +446,6 @@ def generate_owner_hash(creds: Credentials) -> bytes:
un = creds.username
return md5(un.encode())

# Copy & paste from sslaestransport.
def _create_ssl_context(self) -> ssl.SSLContext:
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.set_ciphers(self.CIPHERS)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return context

# Copy & paste from sslaestransport.
async def _get_ssl_context(self) -> ssl.SSLContext:
if not self._ssl_context:
loop = asyncio.get_running_loop()
self._ssl_context = await loop.run_in_executor(
None, self._create_ssl_context
)
return self._ssl_context


class KlapTransportV2(KlapTransport):
"""Implementation of the KLAP encryption protocol with v2 hanshake hashes."""
Expand Down
30 changes: 2 additions & 28 deletions kasa/transports/linkietransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,8 @@

from __future__ import annotations

import asyncio
import base64
import logging
import ssl
from typing import TYPE_CHECKING, cast
from urllib.parse import quote

Expand All @@ -19,6 +17,7 @@
from kasa.transports.xortransport import XorEncryption

from .basetransport import BaseTransport
from .sslcontext import get_ssl_context

_LOGGER = logging.getLogger(__name__)

Expand All @@ -31,20 +30,10 @@ class LinkieTransportV2(BaseTransport):
"""

DEFAULT_PORT: int = 10443
CIPHERS = ":".join(
[
"AES256-GCM-SHA384",
"AES256-SHA256",
"AES128-GCM-SHA256",
"AES128-SHA256",
"AES256-SHA",
]
)

def __init__(self, *, config: DeviceConfig) -> None:
super().__init__(config=config)
self._http_client = HttpClient(config)
self._ssl_context: ssl.SSLContext | None = None
self._app_url = URL(f"https://{self._host}:{self._port}/data/LINKIE2.json")

self._headers = {
Expand Down Expand Up @@ -78,7 +67,7 @@ async def _execute_send(self, request: str) -> dict:
self._app_url,
headers=self._headers,
data=f"content={url_safe_cmd}".encode(),
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)

if TYPE_CHECKING:
Expand Down Expand Up @@ -128,18 +117,3 @@ async def send(self, request: str) -> dict:
raise _RetryableError(
f"Unable to query the device {self._host}:{self._port}: {ex}"
) from ex

async def _get_ssl_context(self) -> ssl.SSLContext:
if not self._ssl_context:
loop = asyncio.get_running_loop()
self._ssl_context = await loop.run_in_executor(
None, self._create_ssl_context
)
return self._ssl_context

def _create_ssl_context(self) -> ssl.SSLContext:
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.set_ciphers(self.CIPHERS)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return context
40 changes: 6 additions & 34 deletions kasa/transports/sslaestransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,10 @@

from __future__ import annotations

import asyncio
import base64
import hashlib
import logging
import secrets
import ssl
from contextlib import suppress
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, cast
Expand All @@ -29,6 +27,7 @@
from ..json import dumps as json_dumps
from ..json import loads as json_loads
from . import AesEncyptionSession, BaseTransport
from .sslcontext import get_ssl_context

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -71,17 +70,6 @@ class SslAesTransport(BaseTransport):
"Accept-Encoding": "gzip, deflate",
"User-Agent": "Tapo CameraClient Android",
}
CIPHERS = ":".join(
[
"ECDHE-RSA-AES128-GCM-SHA256",
"ECDHE-RSA-AES256-GCM-SHA384",
"AES256-GCM-SHA384",
"AES256-SHA256",
"AES128-GCM-SHA256",
"AES128-SHA256",
"AES256-SHA",
]
)
DEFAULT_TIMEOUT = 10

def __init__(
Expand Down Expand Up @@ -113,7 +101,6 @@ def __init__(
self._host_port = f"{self._host}:{self._port}"
self._app_url = URL(f"https://{self._host_port}")
self._token_url: URL | None = None
self._ssl_context: ssl.SSLContext | None = None
ref = str(self._token_url) if self._token_url else str(self._app_url)
self._headers = {
**self.COMMON_HEADERS,
Expand Down Expand Up @@ -210,21 +197,6 @@ def _handle_response_error_code(self, resp_dict: Any, msg: str) -> None:
raise AuthenticationError(msg, error_code=error_code)
raise DeviceError(msg, error_code=error_code)

def _create_ssl_context(self) -> ssl.SSLContext:
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.set_ciphers(self.CIPHERS)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return context

async def _get_ssl_context(self) -> ssl.SSLContext:
if not self._ssl_context:
loop = asyncio.get_running_loop()
self._ssl_context = await loop.run_in_executor(
None, self._create_ssl_context
)
return self._ssl_context

async def send_secure_passthrough(self, request: str) -> dict[str, Any]:
"""Send encrypted message as passthrough."""
if self._state is TransportState.ESTABLISHED and self._token_url:
Expand Down Expand Up @@ -255,7 +227,7 @@ async def send_secure_passthrough(self, request: str) -> dict[str, Any]:
url,
json=passthrough_request_str,
headers=headers,
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)

if TYPE_CHECKING:
Expand Down Expand Up @@ -347,7 +319,7 @@ async def send_unencrypted(self, request: str) -> dict[str, Any]:
url,
json=request,
headers=self._headers,
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)

if status_code != 200:
Expand Down Expand Up @@ -433,7 +405,7 @@ async def try_perform_less_secure_login(self, username: str, password: str) -> b
self._app_url,
json=body,
headers=self._headers,
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)
if status_code != 200:
raise KasaException(
Expand Down Expand Up @@ -477,7 +449,7 @@ async def perform_handshake2(
self._app_url,
json=body,
headers=self._headers,
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)
if status_code != 200:
raise KasaException(
Expand Down Expand Up @@ -660,7 +632,7 @@ async def try_send_handshake1(self, username: str, local_nonce: str) -> dict:
self._app_url,
json=body,
headers=self._headers,
ssl=await self._get_ssl_context(),
ssl=await get_ssl_context(),
)

_LOGGER.debug("Device responded with status %s: %s", status_code, resp_dict)
Expand Down
40 changes: 40 additions & 0 deletions kasa/transports/sslcontext.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
"""Shared SSL context for the transports talking https to local devices.
Not security sensitive: certificates are not verified.
The extra ciphers exist only for compatibility,
as Python's defaults lack suites older firmware needs.
"""

from __future__ import annotations

import asyncio
import ssl
from functools import cache

LEGACY_CIPHERS = [
"AES256-GCM-SHA384",
"AES256-SHA256",
"AES128-GCM-SHA256",
"AES128-SHA256",
"AES256-SHA",
"AES128-SHA",
"ECDHE-RSA-AES128-SHA",
"ECDHE-RSA-AES256-SHA",
]


@cache
def _create_ssl_context() -> ssl.SSLContext:
"""Return the cached context, blocking on first use."""
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
# Append to the defaults so suites newer firmware needs are not dropped.
default = [cipher["name"] for cipher in context.get_ciphers()]
context.set_ciphers(":".join([*default, *LEGACY_CIPHERS]))
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return context


async def get_ssl_context() -> ssl.SSLContext:
"""Return the cached context, created in an executor as that blocks."""
return await asyncio.get_running_loop().run_in_executor(None, _create_ssl_context)
2 changes: 2 additions & 0 deletions kasa/transports/ssltransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
from kasa.json import dumps as json_dumps
from kasa.json import loads as json_loads
from kasa.transports import BaseTransport
from kasa.transports.sslcontext import get_ssl_context

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -151,6 +152,7 @@ async def send_request(self, request: str) -> dict[str, Any]:
url,
json=request,
headers=self.COMMON_HEADERS,
ssl=await get_ssl_context(),
)

if status_code != 200:
Expand Down
6 changes: 0 additions & 6 deletions tests/transports/test_sslaestransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -48,12 +48,6 @@
MOCK_UNENCRYPTED_PASSTHROUGH_STOK = "32charLowerCaseHexStok"


def test_ssl_ciphers_include_modern_tapo_suites() -> None:
"""Keep compatibility with cameras that dropped legacy RSA ciphers."""
assert "ECDHE-RSA-AES128-GCM-SHA256" in SslAesTransport.CIPHERS
assert "ECDHE-RSA-AES256-GCM-SHA384" in SslAesTransport.CIPHERS


@pytest.mark.parametrize(
(
"status_code",
Expand Down
11 changes: 11 additions & 0 deletions tests/transports/test_sslcontext.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
import ssl

from kasa.transports.sslcontext import get_ssl_context


async def test_ssl_context_extends_python_defaults_with_legacy_suites() -> None:
"""Some ciphers are added on top of Python's defaults for compatibility."""
default = {c["name"] for c in ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT).get_ciphers()}
enabled = {c["name"] for c in (await get_ssl_context()).get_ciphers()}
assert default <= enabled
assert {"AES128-GCM-SHA256", "AES256-SHA"} <= enabled
Loading

Back | FazBrowse Home | New Git URL