Chore/2607 update

.github/workflows/python-package.yml

@@ -0,0 +1,96 @@
+# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Python package
+
+on:
+  push:
+    branches: [ "main", "ga" ]
+  pull_request:
+    branches: [ "main", "ga" ]
+
+jobs:
+  lint_and_test:
+
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.9", "3.10", "3.11", "3.13", "3.14"]
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install black pytest
+        python -m pip install ".[dev]"
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+    - name: Lint
+      run: |
+        # stop the build if there are Python syntax errors or undefined names
+        black --quiet --diff --config pyproject.toml --check pywebpush
+        bandit --quiet -c pyproject.toml pywebpush
+    - name: Test with pytest
+      run: |
+        pytest pywebpush
+
+  release-build:
+    runs-on: ubuntu-latest
+    needs: lint_and_test
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.x"
+
+      - name: Build release distributions
+        run: |
+          # NOTE: put your own distribution build steps here.
+          python -m pip install --upgrade pip
+          python -m pip install build
+          python -m build
+
+      - name: Upload distributions
+        uses: actions/upload-artifact@v4
+        with:
+          name: release-dists
+          path: dist/
+
+  pypi-publish:
+    if: ${{ github.event_name == "push" && ( github.ref_name == "main" || startsWith(github.ref, "refs/tags/")) }}
+    runs-on: ubuntu-latest
+    needs:
+      - release-build
+    permissions:
+      # IMPORTANT: this permission is mandatory for trusted publishing
+      id-token: write
+
+    # Dedicated environments with protections for publishing are strongly recommended.
+    # For more information, see: https://docs.github.com/en/actions/deployment/targeting-different-environments/using-environments-for-deployment#deployment-protection-rules
+    environment:
+      name: pypi
+      # OPTIONAL: uncomment and update to include your PyPI project URL in the deployment status:
+      url: https://pypi.org/p/pywebpush
+      #
+      # ALTERNATIVE: if your GitHub Release name is the PyPI project version string
+      # ALTERNATIVE: exactly, uncomment the following line instead:
+      # url: https://pypi.org/project/YOURPROJECT/${{ github.event.release.name }}
+
+    steps:
+      - name: Retrieve release distributions
+        uses: actions/download-artifact@v4
+        with:
+          name: release-dists
+          path: dist/
+
+      # - name: Publish release distributions to PyPI
+      #   uses: pypa/gh-action-pypi-publish@release/v1
+      #   with:
+      #     packages-dir: dist/

.gitignore

@@ -16,6 +16,7 @@ downloads/
 eggs/
 .eggs/
 include/
+ignore/
 local/
 lib/
 lib64/
@@ -26,6 +27,7 @@ wheels/
 share/python-wheels/
 *.egg-info/
 .installed.cfg
+.installed
 *.egg
 MANIFEST
 
@@ -53,6 +55,8 @@ coverage.xml
 .hypothesis/
 .pytest_cache/
 cover/
+ltest/
+.circleci/
 
 # Translations
 *.mo
@@ -125,6 +129,7 @@ celerybeat.pid
 # Environments
 .env
 .venv
+.envrc
 env/
 venv/
 ENV/
@@ -162,4 +167,4 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
-.vscode/
+.vscode/

Makefile

@@ -0,0 +1,27 @@
+# Simple Makefile to help with things like formatting
+# checks and installs
+
+.PHONY: build
+build: .installed
+
+.PHONY: install
+install: .installed
+.installed:
+	pip install ".[dev]"
+	touch .installed
+
+.PHONY: test
+test: .installed
+	pytest
+
+lint: .installed
+	isort --sp pyproject.toml -c pywebpush
+	black --quiet --config pyproject.toml --check --target-version py314 pywebpush
+	bandit --quiet -r -c pyproject.toml pywebpush
+
+format: .installed
+	isort --sp pyproject.toml pywebpush
+	black --quiet --config pyproject.toml --target-version py314 pywebpush
+	bandit --quiet -r -c pyproject.toml pywebpush
+
+

pyproject.toml

@@ -28,7 +28,7 @@ dynamic = ["dependencies"]
 Homepage = "https://github.com/web-push-libs/pywebpush"
 
 [project.optional-dependencies]
-dev = ["black", "mock", "pytest"]
+dev = ["isort", "bandit", "black", "mock", "pytest"]
 
 # create the `pywebpush` helper using `python -m pip install --editable .`
 [project.scripts]
@@ -39,3 +39,29 @@ dependencies = { file = "requirements.txt" }
 
 [tool.setuptools.packages.find]
 include = ["pywebpush*"]
+
+[tool.isort]
+profile = "black"
+skip_gitignore = true
+
+[tool.bandit]
+# skips asserts
+# B101: https://bandit.readthedocs.io/en/latest/plugins/b101_assert_used.html#
+# skip false detect of hardcoded sql
+# B608:https://bandit.readthedocs.io/en/latest/plugins/B608_hardcoded_sql_expressions.html#
+skips = ["B101", "B608"]
+
+[tool.mypy]
+disable_error_code = "attr-defined"
+disallow_untyped_calls = false
+follow_imports = "normal"
+ignore_missing_imports = true
+pretty = true
+show_error_codes = true
+strict_optional = true
+warn_no_return = true
+warn_redundant_casts = true
+warn_return_any = true
+warn_unused_ignores = true
+warn_unreachable = true
+check_untyped_defs = true

pywebpush/__init__.py

@@ -2,23 +2,22 @@
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
-import asyncio
 import base64
 import json
+import logging
 import os
 import time
-import logging
 from copy import deepcopy
-from typing import cast, Union, Dict
+from types import ModuleType
+from typing import Mapping, cast
 from urllib.parse import urlparse
 
 import aiohttp
 import http_ece
 import requests
 from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives import serialization
-from functools import partial
+from cryptography.hazmat.primitives.asymmetric import ec
 from py_vapid import Vapid, Vapid01
 from requests import Response
 
@@ -76,7 +75,9 @@ def get(self, key, default=None):
         except KeyError:
             return default
 
-    def update(self, data) -> None:
+    # Skip mypy check on the following because the declaration is too
+    # abstract
+    def update(self, data: dict) -> None:  # type: ignore
         for key in data:
             self.__setitem__(key, data[key])
 
@@ -116,19 +117,18 @@ class WebPusher:
 
     """
 
-    subscription_info = {}
-    valid_encodings = [
+    subscription_info: Mapping = {}
+    valid_encodings: list[str] = [
         # "aesgcm128",  # this is draft-0, but DO NOT USE.
         "aesgcm",  # draft-httpbis-encryption-encoding-01
         "aes128gcm",  # RFC8188 Standard encoding
     ]
-    verbose = False
+    verbose: bool = False
+    mod_or_session: ModuleType | requests.Session
 
     def __init__(
         self,
-        subscription_info: dict[
-            str, str | bytes | dict[str, str | bytes]
-        ],
+        subscription_info: Mapping,
         requests_session: None | requests.Session = None,
         aiohttp_session: None | aiohttp.client.ClientSession = None,
         verbose: bool = False,
@@ -146,9 +146,9 @@ def __init__(
 
         self.verbose = verbose
         if requests_session is None:
-            self.requests_method = requests
+            self.mod_or_session = requests
         else:
-            self.requests_method = requests_session
+            self.mod_or_session = requests_session
 
         self.aiohttp_session = aiohttp_session
 
@@ -205,7 +205,7 @@ def encode(
         if not self.auth_key or not self.receiver_key:
             raise WebPushException("No keys specified in subscription info")
         self.verb("Encoding data...")
-        salt = None
+        salt: bytes | None = None
         if content_encoding not in self.valid_encodings:
             raise WebPushException(
                 "Invalid content encoding specified. "
@@ -214,7 +214,7 @@ def encode(
         if content_encoding == "aesgcm":
             self.verb("Generating salt for aesgcm...")
             salt = os.urandom(16)
-            logging.debug(f"Salt: {salt}")
+            logging.debug(f"Salt: {salt!r}")
         # The server key is an ephemeral ECDH key used only for this
         # transaction
         server_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
@@ -223,8 +223,6 @@ def encode(
             format=serialization.PublicFormat.UncompressedPoint,
         )
 
-        if isinstance(data, str):
-            data = bytes(data.encode("utf8"))
         if content_encoding == "aes128gcm":
             self.verb("Encrypting to aes128gcm...")
             encrypted = http_ece.encrypt(
@@ -254,7 +252,9 @@ def encode(
                 reply["salt"] = base64.urlsafe_b64encode(salt).strip(b"=")
         return reply
 
-    def as_curl(self, endpoint: str, encoded_data: bytes, headers: dict[str, str]) -> str:
+    def as_curl(
+        self, endpoint: str, encoded_data: bytes, headers: dict[str, str]
+    ) -> str:
         """Return the send as a curl command.
 
         Useful for debugging. This will write out the encoded data to a local
@@ -274,9 +274,7 @@ def as_curl(self, endpoint: str, encoded_data: bytes, headers: dict[str, str]) -
             data = "--data-binary @encrypted.data"
         if "content-length" not in headers:
             self.verb("Generating content-length header...")
-            header_list.append(
-                f'-H "content-length: {len(encoded_data)}" \\ \n'
-            )
+            header_list.append(f'-H "content-length: {len(encoded_data)}" \\ \n')
         return """curl -vX POST {url} \\\n{headers}{data}""".format(
             url=endpoint, headers="".join(header_list), data=data
         )
@@ -303,6 +301,8 @@ def _prepare_send_data(
             headers = dict()
         encoded = CaseInsensitiveDict()
         headers = CaseInsensitiveDict(headers)
+        if isinstance(data, str):
+            data = data.encode()
         if data:
             encoded = self.encode(data, content_encoding)
             if "crypto_key" in encoded:
@@ -354,7 +354,7 @@ def send(self, *args, **kwargs) -> Response | str:
             headers = params["headers"]
             return self.as_curl(endpoint, encoded_data=encoded_data, headers=headers)
 
-        resp = self.requests_method.post(
+        resp = self.mod_or_session.post(
             endpoint,
             timeout=timeout,
             **params,
@@ -363,7 +363,7 @@ def send(self, *args, **kwargs) -> Response | str:
             "\nResponse:\n\tcode: {}\n\tbody: {}\n\theaders: {}",
             resp.status_code,
             resp.text or "Empty",
-            resp.headers or "None"
+            resp.headers or "None",
         )
         return resp
 
@@ -394,9 +394,7 @@ async def send_async(self, *args, **kwargs) -> aiohttp.ClientResponse | str:
 
 
 def webpush(
-    subscription_info: dict[
-        str, str | bytes | dict[str, str | bytes]
-    ],
+    subscription_info: Mapping,
     data: None | str = None,
     vapid_private_key: None | Vapid | str = None,
     vapid_claims: None | dict[str, str | int] = None,
@@ -409,7 +407,7 @@ def webpush(
     requests_session: None | requests.Session = None,
 ) -> str | requests.Response:
     """
-        One call solution to endcode and send `data` to the endpoint
+        One call solution to encode and send `data` to the endpoint
         contained in `subscription_info` using optional VAPID auth headers.
 
         in example:
@@ -513,9 +511,7 @@ def webpush(
 
 
 async def webpush_async(
-    subscription_info: dict[
-        str, str | bytes | dict[str, str | bytes]
-    ],
+    subscription_info: dict[str, str | bytes | dict[str, str | bytes]],
     data: None | str = None,
     vapid_private_key: None | Vapid | str = None,
     vapid_claims: None | dict[str, str | int] = None,