refactor(manager, xray, crypto): improve state serialization and type handling

- Moved state serialization in CoreManager to ensure proper encoding.
- Updated XRayConfig to ensure SNI values are converted to strings.
- Enhanced get_cert_SANs function to return SAN values as strings.
- Added unit tests for get_cert_SANs to verify JSON serializability of returned values.

Author: x0sina

app/core/manager.py

@@ -61,9 +61,8 @@ async def _persist_state(self):
         if not self._kv:
             return
         state = await self._snapshot_state()
-        # State is already serialized by _snapshot_state; just encode
-        state_bytes = json.dumps(state).encode("utf-8")
         try:
+            state_bytes = json.dumps(state).encode("utf-8")
             await self._kv.put(self.STATE_CACHE_KEY, state_bytes)
         except Exception as exc:
             self._logger.warning(f"Failed to persist core state to NATS KV: {exc}")
@@ -80,7 +79,7 @@ async def _load_state_from_cache(self) -> bool:
             # Deserialize state using JSON
             try:
                 cached_state = json.loads(entry.value.decode("utf-8"))
-            except json.JSONDecodeError, UnicodeDecodeError:
+            except (json.JSONDecodeError, UnicodeDecodeError):
                 self._logger.warning("Failed to decode CoreManager state as JSON, ignoring...")
                 return False
 

app/core/xray.py

@@ -181,7 +181,7 @@ def _handle_tls_settings(self, tls_settings: dict, settings: dict, inbound_tag:
         settings["pinnedPeerCertSha256"] = tls_settings.get("pinnedPeerCertSha256", "")
         settings["fp"] = tls_settings.get("fingerprint", "chrome")
         if sni := tls_settings.get("serverName"):
-            settings["sni"].append(sni)
+            settings["sni"].append(str(sni))
         for certificate in tls_settings.get("certificates", []):
             serve_on_node = certificate.pop("serveOnNode", False)
             if serve_on_node:
@@ -218,7 +218,7 @@ def _handle_reality_settings(self, tls_settings: dict, settings: dict, inbound_t
         """Handle Reality security settings."""
         settings["fp"] = tls_settings.get("fingerprint", "chrome")
         settings["tls"] = "reality"
-        settings["sni"] = tls_settings.get("serverNames", [])
+        settings["sni"] = [str(sni) for sni in (tls_settings.get("serverNames") or [])]
 
         pvk = tls_settings.get("privateKey")
         if not pvk:

app/utils/crypto.py

@@ -9,14 +9,14 @@
 from cryptography.hazmat.primitives.asymmetric import x25519
 
 
-def get_cert_SANs(cert: bytes):
+def get_cert_SANs(cert: bytes) -> list[str]:
+    """Return SAN values as strings (IP SANs are ipaddress objects)."""
     cert = x509.load_pem_x509_certificate(cert, default_backend())
-    san_list = []
+    san_list: list[str] = []
     for extension in cert.extensions:
         if isinstance(extension.value, x509.SubjectAlternativeName):
-            san = extension.value
-            for name in san:
-                san_list.append(name.value)
+            for name in extension.value:
+                san_list.append(str(name.value))
     return san_list
 
 

tests/test_cert_sans.py

@@ -0,0 +1,48 @@
+import datetime
+import json
+from ipaddress import ip_address
+
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
+
+from app.utils.crypto import get_cert_SANs
+
+
+def _pem_cert_with_dns_and_ip_sans() -> bytes:
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "example.com")])
+    now = datetime.datetime.now(datetime.UTC)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(name)
+        .issuer_name(name)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - datetime.timedelta(days=1))
+        .not_valid_after(now + datetime.timedelta(days=90))
+        .add_extension(
+            x509.SubjectAlternativeName(
+                [
+                    x509.DNSName("example.com"),
+                    x509.DNSName("www.example.com"),
+                    x509.IPAddress(ip_address("1.2.3.4")),
+                    x509.IPAddress(ip_address("2001:db8::1")),
+                ]
+            ),
+            critical=False,
+        )
+        .sign(key, hashes.SHA256())
+    )
+    return cert.public_bytes(serialization.Encoding.PEM)
+
+
+def test_get_cert_sans_returns_json_serializable_strings() -> None:
+    sans = get_cert_SANs(_pem_cert_with_dns_and_ip_sans())
+    assert all(isinstance(item, str) for item in sans)
+    assert "example.com" in sans
+    assert "www.example.com" in sans
+    assert "1.2.3.4" in sans
+    assert "2001:db8::1" in sans
+    json.dumps({"sni": sans})