Secrets cleanup
This commit is contained in:
parent
34c661e4fa
commit
af3eaa33d4
5 changed files with 194 additions and 58 deletions
|
|
@ -24,6 +24,7 @@ Implementation Notes
|
|||
import gc
|
||||
import os
|
||||
import time
|
||||
import warnings
|
||||
|
||||
from adafruit_fakerequests import Fake_Requests
|
||||
from adafruit_io.adafruit_io import IO_HTTP, AdafruitIO_RequestError
|
||||
|
|
@ -57,11 +58,18 @@ CONTENT_TEXT = const(1)
|
|||
CONTENT_JSON = const(2)
|
||||
CONTENT_IMAGE = const(3)
|
||||
|
||||
OLD_SETTINGS = {
|
||||
OLD_SECRETS = {
|
||||
"ADAFRUIT_AIO_KEY": "aio_key",
|
||||
"ADAFRUIT_AIO_USERNAME": "aio_username",
|
||||
"AIO_KEY": "aio_key",
|
||||
"AIO_USERNAME": "aio_username",
|
||||
"CIRCUITPY_WIFI_SSID": "ssid",
|
||||
"CIRCUITPY_WIFI_PASSWORD": "password",
|
||||
"AIO_USERNAME": "aio_username",
|
||||
"AIO_KEY": "aio_key",
|
||||
}
|
||||
|
||||
OLD_SETTINGS = {
|
||||
"ADAFRUIT_AIO_KEY": "AIO_KEY",
|
||||
"ADAFRUIT_AIO_USERNAME": "AIO_USERNAME",
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -83,12 +91,11 @@ class NetworkBase:
|
|||
:param bool extract_values: If true, single-length fetched values are automatically extracted
|
||||
from lists and tuples. Defaults to ``True``.
|
||||
:param debug: Turn on debug print outs. Defaults to False.
|
||||
:param list secrets_data: An optional list in place of the data contained in the secrets.py file
|
||||
|
||||
"""
|
||||
|
||||
def __init__( # noqa: PLR0912,PLR0913 Too many branches,Too many arguments in function definition
|
||||
self, wifi_module, *, extract_values=True, debug=False, secrets_data=None
|
||||
self, wifi_module, *, extract_values=True, debug=False
|
||||
):
|
||||
self._wifi = wifi_module
|
||||
self._debug = debug
|
||||
|
|
@ -101,11 +108,6 @@ class NetworkBase:
|
|||
]
|
||||
|
||||
self._settings = {}
|
||||
if secrets_data is not None:
|
||||
for key, value in secrets_data.items():
|
||||
if key in OLD_SETTINGS:
|
||||
key = OLD_SETTINGS.get(key) # noqa: PLW2901 `for` loop variable `value` overwritten by assignment target
|
||||
self._settings[key] = value
|
||||
self._wifi_credentials = None
|
||||
|
||||
self.requests = None
|
||||
|
|
@ -120,31 +122,44 @@ class NetworkBase:
|
|||
|
||||
gc.collect()
|
||||
|
||||
def _get_setting(self, setting_name, show_error=True):
|
||||
def _get_setting(self, setting_name):
|
||||
# if setting is has already been found, return it
|
||||
if setting_name in self._settings:
|
||||
return self._settings[setting_name]
|
||||
|
||||
old_setting_name = setting_name
|
||||
# if setting is in settings.toml return it
|
||||
env_value = os.getenv(setting_name)
|
||||
if env_value is not None:
|
||||
self._settings[setting_name] = env_value
|
||||
return env_value
|
||||
|
||||
# if setting old name is in settings.toml return it
|
||||
if setting_name in OLD_SETTINGS:
|
||||
old_setting_name = OLD_SETTINGS.get(setting_name)
|
||||
if os.getenv(setting_name) is not None:
|
||||
return os.getenv(setting_name)
|
||||
env_value = os.getenv(old_setting_name)
|
||||
if env_value is not None:
|
||||
self._settings[setting_name] = env_value
|
||||
return env_value
|
||||
|
||||
try:
|
||||
from secrets import secrets
|
||||
except ImportError:
|
||||
secrets = {}
|
||||
if old_setting_name in secrets.keys():
|
||||
self._settings[setting_name] = secrets[old_setting_name]
|
||||
return self._settings[setting_name]
|
||||
if show_error:
|
||||
if setting_name in ("CIRCUITPY_WIFI_SSID", "CIRCUITPY_WIFI_PASSWORD"):
|
||||
print(
|
||||
"""WiFi settings are kept in settings.toml, please add them there!
|
||||
the secrets dictionary must contain 'CIRCUITPY_WIFI_SSID' and 'CIRCUITPY_WIFI_PASSWORD'
|
||||
at a minimum in order to use network related features"""
|
||||
)
|
||||
else:
|
||||
print(f"{setting_name} not found. Please add this setting to settings.toml.")
|
||||
return None
|
||||
|
||||
# if setting is in legacy secrets.py return it
|
||||
secrets_setting_name = setting_name
|
||||
if setting_name in OLD_SECRETS:
|
||||
# translate common names
|
||||
secrets_setting_name = OLD_SECRETS.get(setting_name)
|
||||
env_value = secrets.get(secrets_setting_name)
|
||||
if env_value is not None:
|
||||
warnings.warn(
|
||||
"The using of `secrets`, is deprecated. Please put your settings in "
|
||||
"settings.toml"
|
||||
)
|
||||
self._settings[setting_name] = env_value
|
||||
return env_value
|
||||
|
||||
return None
|
||||
|
||||
def neo_status(self, value):
|
||||
|
|
@ -206,17 +221,17 @@ class NetworkBase:
|
|||
api_url = None
|
||||
reply = None
|
||||
try:
|
||||
aio_username = self._get_setting("AIO_USERNAME")
|
||||
aio_key = self._get_setting("AIO_KEY")
|
||||
aio_username = self._get_setting("ADAFRUIT_AIO_USERNAME")
|
||||
aio_key = self._get_setting("ADAFRUIT_AIO_KEY")
|
||||
except KeyError:
|
||||
raise KeyError(
|
||||
"\n\nOur time service requires a login/password to rate-limit. "
|
||||
"Please register for a free adafruit.io account and place the user/key "
|
||||
"in your secrets file under 'AIO_USERNAME' and 'AIO_KEY'"
|
||||
"Please register for a free adafruit.io account and place the user/key in "
|
||||
"your settings.toml file under 'ADAFRUIT_AIO_USERNAME' and 'ADAFRUIT_AIO_KEY'"
|
||||
) from KeyError
|
||||
|
||||
if location is None:
|
||||
location = self._get_setting("timezone", False)
|
||||
location = self._get_setting("timezone")
|
||||
if location:
|
||||
print("Getting time for timezone", location)
|
||||
api_url = (TIME_SERVICE + "&tz=%s") % (aio_username, aio_key, location)
|
||||
|
|
@ -242,7 +257,8 @@ class NetworkBase:
|
|||
reply = response.text
|
||||
except KeyError:
|
||||
raise KeyError(
|
||||
"Was unable to lookup the time, try setting secrets['timezone'] according to http://worldtimeapi.org/timezones"
|
||||
"Was unable to lookup the time, try setting 'timezone' in your settings.toml"
|
||||
"according to http://worldtimeapi.org/timezones"
|
||||
) from KeyError
|
||||
# now clean up
|
||||
response.close()
|
||||
|
|
@ -346,7 +362,7 @@ class NetworkBase:
|
|||
|
||||
def connect(self, max_attempts=10):
|
||||
"""
|
||||
Connect to WiFi using the settings found in secrets.py
|
||||
Connect to WiFi using the settings found in settings.toml
|
||||
|
||||
:param max_attempts: The maximum number of attempts to connect to WiFi before
|
||||
failing or use None to disable. Defaults to 10.
|
||||
|
|
@ -361,7 +377,7 @@ class NetworkBase:
|
|||
}
|
||||
]
|
||||
|
||||
networks = self._get_setting("networks", False)
|
||||
networks = self._get_setting("networks")
|
||||
if networks is not None:
|
||||
if isinstance(networks, (list, tuple)):
|
||||
self._wifi_credentials = networks
|
||||
|
|
@ -370,14 +386,14 @@ class NetworkBase:
|
|||
"'networks' must be a list/tuple of dicts of 'ssid' and 'password'"
|
||||
)
|
||||
|
||||
for secret_entry in self._wifi_credentials:
|
||||
for credentials in self._wifi_credentials:
|
||||
self._wifi.neo_status(STATUS_CONNECTING)
|
||||
attempt = 1
|
||||
|
||||
while not self._wifi.is_connected:
|
||||
# secrets dictionary must contain 'ssid' and 'password' at a minimum
|
||||
print("Connecting to AP", secret_entry["ssid"])
|
||||
if secret_entry["ssid"] == "CHANGE ME" or secret_entry["password"] == "CHANGE ME":
|
||||
# credentials must contain 'CIRCUITPY_WIFI_SSID' and 'CIRCUITPY_WIFI_PASSWORD'
|
||||
print("Connecting to AP", credentials["ssid"])
|
||||
if credentials["ssid"] == "CHANGE ME" or credentials["password"] == "CHANGE ME":
|
||||
change_me = "\n" + "*" * 45
|
||||
change_me += "\nPlease update the 'settings.toml' file on your\n"
|
||||
change_me += "CIRCUITPY drive to include your local WiFi\n"
|
||||
|
|
@ -387,7 +403,7 @@ class NetworkBase:
|
|||
raise OSError(change_me)
|
||||
self._wifi.neo_status(STATUS_NO_CONNECTION) # red = not connected
|
||||
try:
|
||||
self._wifi.connect(secret_entry["ssid"], secret_entry["password"])
|
||||
self._wifi.connect(credentials["ssid"], credentials["password"])
|
||||
self.requests = self._wifi.requests
|
||||
self._wifi.neo_status(STATUS_CONNECTED)
|
||||
break
|
||||
|
|
@ -412,11 +428,11 @@ class NetworkBase:
|
|||
self.connect()
|
||||
|
||||
try:
|
||||
aio_username = self._get_setting("AIO_USERNAME")
|
||||
aio_key = self._get_setting("AIO_KEY")
|
||||
aio_username = self._get_setting("ADAFRUIT_AIO_USERNAME")
|
||||
aio_key = self._get_setting("ADAFRUIT_AIO_KEY")
|
||||
except KeyError:
|
||||
raise KeyError(
|
||||
"Adafruit IO secrets are kept in secrets.py, please add them there!\n\n"
|
||||
"Adafruit IO settings are kept in settings.toml, please add them there!\n\n"
|
||||
) from KeyError
|
||||
|
||||
self._io_client = IO_HTTP(aio_username, aio_key, self._wifi.requests)
|
||||
|
|
|
|||
|
|
@ -70,7 +70,6 @@ class WiFi:
|
|||
|
||||
if self.esp.is_connected:
|
||||
self._set_requests()
|
||||
self._manager = None
|
||||
|
||||
gc.collect()
|
||||
|
||||
|
|
@ -81,9 +80,9 @@ class WiFi:
|
|||
|
||||
def connect(self, ssid, password):
|
||||
"""
|
||||
Connect to WiFi using the settings found in secrets.py
|
||||
Connect to WiFi using the settings found in settings.toml
|
||||
"""
|
||||
self.esp.connect({"ssid": ssid, "password": password})
|
||||
self.esp.connect(ssid, password)
|
||||
self._set_requests()
|
||||
|
||||
def neo_status(self, value):
|
||||
|
|
@ -95,14 +94,6 @@ class WiFi:
|
|||
if self.neopix:
|
||||
self.neopix.fill(value)
|
||||
|
||||
def manager(self, secrets):
|
||||
"""Initialize the WiFi Manager if it hasn't been cached and return it"""
|
||||
if self._manager is None:
|
||||
self._manager = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(
|
||||
self.esp, secrets, None
|
||||
)
|
||||
return self._manager
|
||||
|
||||
@property
|
||||
def is_connected(self):
|
||||
"""Return whether we are connected."""
|
||||
|
|
|
|||
|
|
@ -6,12 +6,12 @@ NOTE: This PortalBase library is intended to be subclassed by other libraries ra
|
|||
used directly by end users. This example shows one such usage with the PyPortal library.
|
||||
See MatrixPortal, MagTag, and PyPortal libraries for more examples.
|
||||
"""
|
||||
# NOTE: Make sure you've created your secrets.py file before running this example
|
||||
# https://learn.adafruit.com/adafruit-pyportal/internet-connect#whats-a-secrets-file-17-2
|
||||
# NOTE: Make sure you've created your settings.toml file before running this example
|
||||
# https://learn.adafruit.com/adafruit-pyportal/create-your-settings-toml-file
|
||||
|
||||
import board
|
||||
import displayio
|
||||
from adafruit_pyportal import PyPortal
|
||||
from displayio import CIRCUITPYTHON_TERMINAL
|
||||
|
||||
# Set a data source URL
|
||||
TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html"
|
||||
|
|
@ -20,7 +20,7 @@ TEXT_URL = "http://wifitest.adafruit.com/testwifi/index.html"
|
|||
pyportal = PyPortal(url=TEXT_URL, status_neopixel=board.NEOPIXEL)
|
||||
|
||||
# Set display to show REPL
|
||||
board.DISPLAY.root_group = displayio.CIRCUITPYTHON_TERMINAL
|
||||
board.DISPLAY.root_group = CIRCUITPYTHON_TERMINAL
|
||||
|
||||
# Go get that data
|
||||
print("Fetching text from", TEXT_URL)
|
||||
|
|
|
|||
|
|
@ -1,3 +1,5 @@
|
|||
# SPDX-FileCopyrightText: 2022 Alec Delaney, for Adafruit Industries
|
||||
#
|
||||
# SPDX-License-Identifier: Unlicense
|
||||
|
||||
pytest
|
||||
|
|
|
|||
127
tests/test_get_settings.py
Normal file
127
tests/test_get_settings.py
Normal file
|
|
@ -0,0 +1,127 @@
|
|||
# SPDX-FileCopyrightText: 2025 Justin Myers
|
||||
#
|
||||
# SPDX-License-Identifier: Unlicense
|
||||
|
||||
import os
|
||||
import sys
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from adafruit_portalbase.network import NetworkBase
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def secrets():
|
||||
sys.modules["secrets.secrets"] = {
|
||||
"aio_key": "secret_aio_key",
|
||||
"aio_username": "secret_aio_username",
|
||||
"password": "secret_password",
|
||||
"ssid": "secret_ssid",
|
||||
"timezone": "secret_timezone",
|
||||
"fallback_test": "secret_fallback_test",
|
||||
}
|
||||
yield
|
||||
del sys.modules["secrets.secrets"]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings_toml_current(monkeypatch):
|
||||
monkeypatch.setenv("ADAFRUIT_AIO_KEY", "settings_current_aio_key")
|
||||
monkeypatch.setenv("ADAFRUIT_AIO_USERNAME", "settings_current_aio_username")
|
||||
monkeypatch.setenv("CIRCUITPY_WIFI_PASSWORD", "settings_current_password")
|
||||
monkeypatch.setenv("CIRCUITPY_WIFI_SSID", "settings_current_ssid")
|
||||
monkeypatch.setenv("timezone", "settings_current_timezone")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings_toml_old(monkeypatch):
|
||||
monkeypatch.setenv("AIO_KEY", "settings_old_aio_key")
|
||||
monkeypatch.setenv("AIO_USERNAME", "settings_old_aio_username")
|
||||
monkeypatch.setenv("CIRCUITPY_WIFI_PASSWORD", "settings_old_password")
|
||||
monkeypatch.setenv("CIRCUITPY_WIFI_SSID", "settings_old_ssid")
|
||||
monkeypatch.setenv("timezone", "settings_old_timezone")
|
||||
|
||||
|
||||
def test_get_setting_does_not_exist():
|
||||
network = NetworkBase(None)
|
||||
assert network._get_setting("test") == None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("key", "value"),
|
||||
(
|
||||
("ADAFRUIT_AIO_KEY", "secret_aio_key"),
|
||||
("ADAFRUIT_AIO_USERNAME", "secret_aio_username"),
|
||||
("AIO_KEY", "secret_aio_key"),
|
||||
("AIO_USERNAME", "secret_aio_username"),
|
||||
("CIRCUITPY_WIFI_PASSWORD", "secret_password"),
|
||||
("CIRCUITPY_WIFI_SSID", "secret_ssid"),
|
||||
("timezone", "secret_timezone"),
|
||||
("not_found", None),
|
||||
),
|
||||
)
|
||||
def test_get_setting_in_secrets(secrets, key, value):
|
||||
network = NetworkBase(None)
|
||||
with mock.patch("adafruit_portalbase.network.warnings.warn") as mock_warnings:
|
||||
assert network._get_setting(key) == value
|
||||
if value:
|
||||
mock_warnings.assert_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("key", "value"),
|
||||
(
|
||||
("ADAFRUIT_AIO_KEY", "settings_current_aio_key"),
|
||||
("ADAFRUIT_AIO_USERNAME", "settings_current_aio_username"),
|
||||
("CIRCUITPY_WIFI_PASSWORD", "settings_current_password"),
|
||||
("CIRCUITPY_WIFI_SSID", "settings_current_ssid"),
|
||||
("timezone", "settings_current_timezone"),
|
||||
("not_found", None),
|
||||
),
|
||||
)
|
||||
def test_get_setting_in_settings_current(settings_toml_current, key, value):
|
||||
network = NetworkBase(None)
|
||||
with mock.patch("adafruit_portalbase.network.warnings.warn") as mock_warnings:
|
||||
assert network._get_setting(key) == value
|
||||
mock_warnings.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("key", "value"),
|
||||
(
|
||||
("ADAFRUIT_AIO_KEY", "settings_old_aio_key"),
|
||||
("ADAFRUIT_AIO_USERNAME", "settings_old_aio_username"),
|
||||
("CIRCUITPY_WIFI_PASSWORD", "settings_old_password"),
|
||||
("CIRCUITPY_WIFI_SSID", "settings_old_ssid"),
|
||||
("timezone", "settings_old_timezone"),
|
||||
("not_found", None),
|
||||
),
|
||||
)
|
||||
def test_get_setting_in_settings_old(settings_toml_old, key, value):
|
||||
network = NetworkBase(None)
|
||||
with mock.patch("adafruit_portalbase.network.warnings.warn") as mock_warnings:
|
||||
assert network._get_setting(key) == value
|
||||
mock_warnings.assert_not_called()
|
||||
if key in ["ADAFRUIT_AIO_KEY", "ADAFRUIT_AIO_USERNAME"]:
|
||||
assert os.getenv(key) is None
|
||||
|
||||
|
||||
def test_fallback(secrets, settings_toml_current):
|
||||
network = NetworkBase(None)
|
||||
with mock.patch("adafruit_portalbase.network.warnings.warn") as mock_warnings:
|
||||
assert network._get_setting("ADAFRUIT_AIO_KEY") == "settings_current_aio_key"
|
||||
mock_warnings.assert_not_called()
|
||||
with mock.patch("adafruit_portalbase.network.warnings.warn") as mock_warnings:
|
||||
assert network._get_setting("fallback_test") == "secret_fallback_test"
|
||||
mock_warnings.assert_called()
|
||||
|
||||
|
||||
def test_value_stored(settings_toml_current):
|
||||
network = NetworkBase(None)
|
||||
with mock.patch("os.getenv", return_value="test") as mock_getenv:
|
||||
assert network._get_setting("ADAFRUIT_AIO_KEY") == "test"
|
||||
mock_getenv.assert_called()
|
||||
with mock.patch("os.getenv", return_value="test") as mock_getenv:
|
||||
assert network._get_setting("ADAFRUIT_AIO_KEY") == "test"
|
||||
mock_getenv.assert_not_called()
|
||||
Loading…
Reference in a new issue