chore: Safe fixes for Ruff 0.16

.github/workflows/ruff_lint.yml

@@ -23,7 +23,6 @@ jobs:
       - name: Install and run Ruff linter
         uses: astral-sh/ruff-action@278981a28ce3188b1e39527901f38254bf3aac89 # v4.1.0
         with:
-          version: 0.15.22
           src: >-
             src/main.py
             src/lib/aes.py

src/lib/aes.py

@@ -1,6 +1,8 @@
 import binascii
+
 import ucryptolib
 
+
 def aes_decrypt(encryptedText, selfEncryptionKey):
     ciphertext = binascii.a2b_base64(encryptedText)
     key = binascii.a2b_base64(selfEncryptionKey)

src/lib/iv_nonce.py

@@ -2,8 +2,10 @@
 """
 
 import random
+
 import ubinascii
 
+
 class IVNonce:
 
     def _randbytes(self, nbytes):

src/lib/logging.py

@@ -1,5 +1,6 @@
 # from https://github.com/micropython/micropython-lib/blob/master/logging/logging.py
 import sys
+
 import utime
 
 CRITICAL = 50
@@ -36,17 +37,17 @@ def setFormatter(self, fmtr):
 class Logger:
 
     level = NOTSET
-    handlers = []
     record = LogRecord()
 
     def __init__(self, name):
         self.name = name
+        self.handlers = []
 
     def _level_str(self, level):
         levelfromdict = _level_dict.get(level)
         if levelfromdict is not None:
             return levelfromdict
-        return "LVL%s" % level
+        return f"LVL{level}"
 
     def setLevel(self, level):
         self.level = level
@@ -65,7 +66,7 @@ def log(self, level, msg, *args):
                         msg = msg % args
                     else:
                         msg = msg.format(*args)
-                except Exception:
+                except ValueError:
                     msg = msg + '--BAD LOG FORMAT--'
             if self.handlers:
                 d = self.record.__dict__

src/lib/ntp_client.py

@@ -1,7 +1,8 @@
-import machine
 import struct
 import time
-import usocket as socket # type: ignore
+
+import machine
+import usocket as socket  # type: ignore
 
 # Substract 1 hour, so we get + 1 hour
 TIMESTAMP_DELTA = 2208988800 - 3600*1 # epoch time - 1 hour
@@ -18,7 +19,7 @@ def sync_time():
         if str(e) == '119': # For non-Blocking sockets 119 is EINPROGRESS
             print("In Progress")
         else:
-            raise e
+            raise
 
     s.sendto(NTP_QUERY, a)
     data = s.recv(48)

src/lib/pem_service.py

@@ -1,6 +1,8 @@
 import io
-import ubinascii
+
 import uasn1
+import ubinascii
+
 
 def read_pem(input_data):
     """Read PEM formatted input."""
@@ -45,7 +47,7 @@ def strid(id):
     elif id == uasn1.Set:
         s = 'SET'
     else:
-        s = '%#02x' % id
+        s = f'{id:#02x}'
     return s
 
 def strclass(id):
@@ -59,12 +61,12 @@ def strclass(id):
     elif id == uasn1.ClassPrivate:
         s = 'PRIVATE'
     else:
-        raise ValueError('Illegal class: %#02x' % id)
+        raise ValueError(f'Illegal class: {id:#02x}')
     return s
 
 def strtag(tag):
     """Return a string represenation of an ASN.1 tag."""
-    return '[%s] %s' % (strid(tag[0]), strclass(tag[2]))
+    return f'[{strid(tag[0])}] {strclass(tag[2])}'
 
 def prettyprint(input_data, output, indent=0):
     """Pretty print ASN.1 data."""
@@ -73,22 +75,19 @@ def prettyprint(input_data, output, indent=0):
         if tag[1] == uasn1.TypePrimitive:
             tag, value = input_data.read()
             output.write(' ' * indent)
-            output.write('[%s] %s (value %s)' %
-                        (strclass(tag[2]), strid(tag[0]), repr(value)))
+            output.write(f'[{strclass(tag[2])}] {strid(tag[0])} (value {value!r})')
             output.write('\n')
         elif tag[1] == uasn1.TypeConstructed:
             output.write(' ' * indent)
-            output.write('[%s] %s:\n' % (strclass(tag[2]), strid(tag[0])))
+            output.write(f'[{strclass(tag[2])}] {strid(tag[0])}:\n')
             input_data.enter()
             prettyprint(input_data, output, indent+2)
             input_data.leave()
 
 def get_pem_parameters(pem):
     formatted_pem = format_pem(pem)
     input_data = read_pem(formatted_pem)
-    data = []
-    for line in input_data:
-        data.append(line)
+    data = input_data.copy()
     if isinstance(data[0], str):
         data = b''.join(data)
     elif isinstance(data[0], int):
@@ -117,9 +116,7 @@ def get_pem_parameters(pem):
 def get_pub_parameters(pkcs1):
     formatted_pkcs1 = format_pub(pkcs1)
     input_data = read_pem(formatted_pkcs1)
-    data = []
-    for line in input_data:
-        data.append(line)
+    data = input_data.copy()
     if isinstance(data[0], str):
         data = b''.join(data)
     elif isinstance(data[0], int):
@@ -152,9 +149,7 @@ def get_pub_parameters(pkcs1):
 def get_pem_key(pkcs8):
     formatted_pkcs8 = format_pem(pkcs8)
     input_data = read_pem(formatted_pkcs8)
-    data = []
-    for line in input_data:
-        data.append(line)
+    data = input_data.copy()
     if isinstance(data[0], str):
         data = b''.join(data)
     elif isinstance(data[0], int):

src/lib/uasn1.py

@@ -6,9 +6,10 @@
 # uASN1 is copyright (c) 2007-2021 by the uASN1 authors. See the
 # file "AUTHORS" for a complete overview.
 
-import ubinascii as binascii
 import re
 
+import ubinascii as binascii
+
 Boolean = 0x01
 Integer = 0x02
 BitString = 0x03
@@ -31,7 +32,7 @@ class Error(Exception):
     """ASN1 error"""
 
 
-class Encoder(object):
+class Encoder:
     """A ASN.1 encoder. Uses DER encoding."""
 
     def __init__(self):
@@ -195,7 +196,7 @@ def _encode_integer(self, value):
     def _encode_octet_string(self, value):
         """Encode an octetstring."""
         # Use the primitive encoding
-        assert isinstance(value, str) or isinstance(value, bytes)
+        assert isinstance(value, (str, bytes))
         if isinstance(value, str):
             return value.encode('utf-8')
         else:
@@ -226,7 +227,7 @@ def _encode_object_identifier(self, oid):
         return bytes(result)
 
 
-class Decoder(object):
+class Decoder:
     """A ASN.1 decoder. Understands BER (and DER which is a subset)."""
 
     def __init__(self):
@@ -270,7 +271,7 @@ def enter(self):
         """Enter a constructed tag."""
         if self.m_stack is None:
             raise Error('No input selected. Call start() first.')
-        nr, typ, cls = self.peek()
+        _nr, typ, _cls = self.peek()
         if typ != TypeConstructed:
             raise Error('Cannot enter a non-constructed tag.')
         length = self._read_length()
@@ -291,9 +292,7 @@ def _decode_boolean(self, bytes_data):
         """Decode a boolean value."""
         if len(bytes_data) != 1:
             raise Error('ASN1 syntax error')
-        if bytes_data[0] == 0:
-            return False
-        return True
+        return bytes_data[0] != 0
 
     def _read_tag(self):
         """Read a tag from the input."""
@@ -424,7 +423,6 @@ def _decode_null(self, bytes_data):
         """Decode a Null value."""
         if len(bytes_data) != 0:
             raise Error('ASN1 syntax error')
-        return None
 
     def _decode_object_identifier(self, bytes_data):
         """Decode an object identifier."""

src/main.py

@@ -1,13 +1,15 @@
-#import _thread
+import _thread
 import sys
+
 # Needed when running on Linux to find imports in lib directory
 if sys.platform == 'linux':
     sys.path.append('./lib')
-import atclient
 import logging
 import os
 import sys
 import time
+
+import atclient
 import ujson as json
 
 log=logging.getLogger(__name__)
@@ -53,9 +55,9 @@ def write_keys(ssid, password, atSign, atRecipient):
     """Write extracted keys into settings.json"""
     log.info("Writing keys")
     from aes import aes_decrypt
-    from pem_service import get_pem_parameters, get_pem_key
-    (aesEncryptPrivateKey, aesEncryptPublicKey, aesPkamPrivateKey,
-            aesPkamPublicKey, selfEncryptionKey) = read_keys(atSign)
+    from pem_service import get_pem_key, get_pem_parameters
+    (aesEncryptPrivateKey, _aesEncryptPublicKey, aesPkamPrivateKey,
+            _aesPkamPublicKey, selfEncryptionKey) = read_keys(atSign)
     pkamPrivateKey = aes_decrypt(aesPkamPrivateKey, selfEncryptionKey)
     encryptPrivateKey = aes_decrypt(aesEncryptPrivateKey, selfEncryptionKey)
     pkamKey = get_pem_parameters(get_pem_key(pkamPrivateKey))
@@ -78,7 +80,7 @@ def main():
         # Transfer keys from atKeys file to settings
         write_keys(ssid,password,atSign,atRecipient)
     if sys.platform != 'linux':
-        import network # type: ignore
+        import network  # type: ignore
         from ntp_client import sync_time
         wlan = network.WLAN(network.STA_IF)  # type: ignore
         wlan.active(True)
@@ -92,7 +94,6 @@ def main():
     connected = False
 
     while True:
-        global lock
         print("Welcome! What would you like to do?\n"
             "\t1) Change recipient atSign (presently " + atRecipient + ")\n"
             "\t2) Connect to " + atSign + "\n"
@@ -118,9 +119,8 @@ def main():
         elif int(opt) == 3:
             if connected:
                 # init second thread to read from socket (monitor)
-                global monitoring
-                monitoring = True
-                #read_thread = _thread.start_new_thread(atc.attalk_recv, ())
+                atclient.monitoring = True
+                _thread.start_new_thread(atc.attalk_recv, ())
                 print('To return to menu type: /exit\n')
                 while True:
                     # print(atSign+":",end='\r')
@@ -130,9 +130,9 @@ def main():
                         break
                     atc.attalk_send(msg=msg)
                 # stop second thread
-                lock.acquire(1)
-                monitoring = False
-                lock.release()
+                atclient.lock.acquire(1)
+                atclient.monitoring = False
+                atclient.lock.release()
                 # join method does not exist in _thread
             else:
                 print("You must connect to " + atSign + " before continuing")