Add Elliptic Curve (EC) Authentication (#21)

* Adds a single public_key_pem column to the database.
* Add --auth-list-public-keys to docker entrypoint's superlink cmd
* Add --auth-list-public-keys to superlink subprocess cmd
* Add blossomfile generation
* Handle custom csv file format support
* Convert PEM-formatted key to ssh format
* Fix tests and dependencies
* Fix test_rebuild_overwrites_existing_file
* Add requirements.txt for custom HF space install

alembic/versions/e7169fe29ea1_add_public_key_pem_to_request_table.py

@@ -0,0 +1,33 @@
+"""Add public_key_pem to Request table.
+
+Revision ID: e7169fe29ea1
+Revises: 0b01b44ab005
+Create Date: 2025-10-09 13:23:30.141015
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "e7169fe29ea1"
+down_revision: Union[str, Sequence[str], None] = "0b01b44ab005"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column("requests", sa.Column("public_key_pem", sa.String(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("requests", "public_key_pem")
+    # ### end Alembic commands ###

blossomtune_gradio/auth_keys.py

@@ -1,39 +1,69 @@
 import os
-import csv
 import logging
 from typing import List, Tuple
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives import serialization
 
+# Configure logging for the module
 log = logging.getLogger(__name__)
 
 
+def _sanitize_key(participant_id: str, key_str: str) -> str | None:
+    """
+    Inspects a key string and converts it to the required OpenSSH format if necessary.
+    This provides resilience against old, PEM-formatted keys in the database.
+    """
+    if not key_str:
+        return None
+    # If the key is already in the correct OpenSSH format, return it as is.
+    if key_str.startswith("ecdsa-sha2-nistp384"):
+        return key_str
+    # If the key is in the old PEM format, attempt to convert it.
+    if "-----BEGIN PUBLIC KEY-----" in key_str:
+        log.warning(
+            f"Found PEM-formatted key for participant {participant_id}. Converting to OpenSSH."
+        )
+        try:
+            public_key = serialization.load_pem_public_key(key_str.encode("utf-8"))
+            key_body = public_key.public_bytes(
+                encoding=serialization.Encoding.OpenSSH,
+                format=serialization.PublicFormat.OpenSSH,
+            ).decode("utf-8")
+            # Re-add the participant_id as a comment to conform to the 3-part format.
+            return f"{key_body} {participant_id}"
+        except Exception as e:
+            log.error(f"Could not convert PEM key for {participant_id}: {e}")
+            return None
+    # If the key format is unknown, log an error and skip it.
+    log.error(f"Unknown public key format for participant {participant_id}. Skipping.")
+    return None
+
+
 def rebuild_authorized_keys_csv(
     key_dir: str, authorized_participants: List[Tuple[str, str]]
 ):
     """
-    Overwrites the public key CSV with a fresh list from a trusted source.
-
-    This function should be called before starting the Flower SuperLink to ensure
-    the list of authorized nodes is always in sync with the database.
-
-    Args:
-        key_dir (str): The directory where the CSV will be stored.
-        authorized_participants (List[Tuple[str, str]]): A list of tuples,
-            where each tuple contains (participant_id, public_key_pem).
+    Overwrites the public key file with a fresh list from a trusted source,
+    using the specific single-line, comma-separated format expected by Flower.
     """
     csv_path = os.path.join(key_dir, "authorized_supernodes.csv")
-    log.info(f"Rebuilding authorized keys CSV at: {csv_path}")
+    log.info(f"Rebuilding authorized keys file at: {csv_path}")
 
-    with open(csv_path, "w", newline="") as f:
-        writer = csv.writer(f)
-        writer.writerow(["participant_id", "public_key_pem"])
-        for participant_id, public_key_pem in authorized_participants:
-            writer.writerow([participant_id, public_key_pem])
+    # Sanitize each key before adding it to the list.
+    public_keys = [
+        sanitized_key
+        for p_id, key_string in authorized_participants
+        if (sanitized_key := _sanitize_key(p_id, key_string)) is not None
+    ]
 
-    log.info(
-        f"Successfully rebuilt {csv_path} with {len(authorized_participants)} keys."
-    )
+    # Join all valid public keys into a single comma-separated string.
+    content = ",".join(public_keys)
+
+    # Write the single line to the file, followed by a newline.
+    with open(csv_path, "w") as f:
+        f.write(content + "\n")
+
+    log.info(f"Successfully rebuilt {csv_path} with {len(public_keys)} keys.")
 
 
 class AuthKeyGenerator:
@@ -43,12 +73,6 @@ class AuthKeyGenerator:
     """
 
     def __init__(self, key_dir: str = "keys"):
-        """
-        Initializes the generator and ensures the key directory exists.
-
-        Args:
-            key_dir (str): The directory where key files will be stored.
-        """
         self.key_dir = key_dir
         os.makedirs(self.key_dir, exist_ok=True)
         log.info(f"Authentication key directory set to: {self.key_dir}")
@@ -70,24 +94,17 @@ class AuthKeyGenerator:
                     encryption_algorithm=serialization.NoEncryption(),
                 )
             )
-        # Set file permissions to read/write for owner only (600)
         os.chmod(priv_key_path, 0o600)
         log.info(f"Private key for {participant_id} saved securely to {priv_key_path}")
         return priv_key_path
 
     def _save_public_key_file(
-        self, private_key: ec.EllipticCurvePrivateKey, participant_id: str
+        self, public_key_ssh_string: str, participant_id: str
     ) -> str:
-        """Saves the public key to a .pub file."""
-        public_key = private_key.public_key()
+        """Saves the full OpenSSH public key string to a .pub file."""
         pub_key_path = os.path.join(self.key_dir, f"{participant_id}.pub")
-        with open(pub_key_path, "wb") as f:
-            f.write(
-                public_key.public_bytes(
-                    encoding=serialization.Encoding.PEM,
-                    format=serialization.PublicFormat.SubjectPublicKeyInfo,
-                )
-            )
+        with open(pub_key_path, "w") as f:
+            f.write(public_key_ssh_string)
         log.info(f"Public key for {participant_id} saved to {pub_key_path}")
         return pub_key_path
 
@@ -95,26 +112,25 @@ class AuthKeyGenerator:
         """
         Generates and saves a new EC key pair for a participant.
 
-        This is the main public method to call when a new participant is approved.
-
-        Args:
-            participant_id (str): A unique identifier for the participant.
-
         Returns:
             A tuple containing:
             - The file path to the generated private key.
             - The file path to the generated public key.
-            - The public key as a PEM-encoded string (for database storage).
+            - The public key as a single-line OpenSSH string with a comment.
         """
         private_key = self._generate_key_pair()
         public_key = private_key.public_key()
 
-        priv_key_path = self._save_private_key(private_key, participant_id)
-        pub_key_path = self._save_public_key_file(private_key, participant_id)
-
-        public_key_pem = public_key.public_bytes(
-            encoding=serialization.Encoding.PEM,
-            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        # Generate the base OpenSSH key string (type and key data)
+        key_body = public_key.public_bytes(
+            encoding=serialization.Encoding.OpenSSH,
+            format=serialization.PublicFormat.OpenSSH,
         ).decode("utf-8")
 
-        return (priv_key_path, pub_key_path, public_key_pem)
+        # Append the participant_id as a comment, creating the full 3-part key.
+        public_key_ssh_string = f"{key_body} {participant_id}"
+
+        priv_key_path = self._save_private_key(private_key, participant_id)
+        pub_key_path = self._save_public_key_file(public_key_ssh_string, participant_id)
+
+        return (priv_key_path, pub_key_path, public_key_ssh_string)

blossomtune_gradio/blossomfile.py

@@ -39,7 +39,7 @@ def create_blossomfile(
         The full path to the generated .blossomfile.
     """
     os.makedirs(output_dir, exist_ok=True)
-    blossomfile_path = os.path.join(output_dir, f"{participant_id}.blossomfile")
+    blossomfile_path = os.path.join(output_dir, f"{participant_id}-blossomfile.zip")
     log.info(f"Creating Blossomfile for {participant_id} at {blossomfile_path}")
 
     # 1. Create the blossom.json configuration data

blossomtune_gradio/config.py

@@ -22,7 +22,7 @@ SMTP_REQUIRE_TLS = util.strtobool(os.getenv("SMTP_REQUIRE_TLS", "false"))
 SMTP_USER = os.getenv("SMTP_USER", "")
 SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "")
 EMAIL_PROVIDER = os.getenv("EMAIL_PROVIDER", "smtp")
-SUPERLINK_HOST = os.getenv("SUPERLINK_HOST", "127.0.0.1:9092")
+SUPERLINK_HOST = os.getenv("SUPERLINK_HOST", "127.0.0.1")
 SUPERLINK_PORT = int(os.getenv("SUPERLINK_PORT", 9092))
 SUPERLINK_CONTROL_API_PORT = int(os.getenv("SUPERLINK_CONTROL_API_PORT", 9093))
 SUPERLINK_MODE = os.getenv("SUPERLINK_MODE", "internal").lower()  # Or external
@@ -50,6 +50,20 @@ BLOSSOMTUNE_TLS_CA_CERTFILE = os.path.join(BLOSSOMTUNE_TLS_CERT_PATH, "ca.crt")
 BLOSSOMTUNE_TLS_CERTFILE = os.path.join(BLOSSOMTUNE_TLS_CERT_PATH, "server.pem")
 BLOSSOMTUNE_TLS_KEYFILE = os.path.join(BLOSSOMTUNE_TLS_CERT_PATH, "server.key")
 
+# EC Auth - Keys
+AUTH_KEYS_DIR = os.getenv(
+    "AUTH_KEYS_DIR",
+    "/data/keys/"
+    if os.path.isdir("/data/keys")
+    else os.path.join(PROJECT_PATH, "./data/keys/"),
+)
+
+# EC Auth - CSV File
+AUTH_KEYS_CSV_PATH = os.getenv(
+    "AUTH_KEYS_CSV_PATH",
+    os.path.join(AUTH_KEYS_DIR, "authorized_supernodes.csv"),
+)
+
 # Flower Apps
 FLOWER_APPS = os.getenv("FLOWER_APPS", ["flower_apps.quickstart_huggingface"])
 FLOWER_APPS = (

blossomtune_gradio/database.py

@@ -29,6 +29,7 @@ class Request(Base):
     hf_handle = Column(String, nullable=True)
     activation_code = Column(String, nullable=True)
     is_activated = Column(Integer, nullable=False, default=0)
+    public_key_pem = Column(String(), nullable=True)
 
     def __repr__(self):
         return (

blossomtune_gradio/federation.py

@@ -1,11 +1,15 @@
+import os
 import string
 import secrets
+import tempfile
 
 from blossomtune_gradio import config as cfg
 from blossomtune_gradio import mail
 from blossomtune_gradio import util
 from blossomtune_gradio.settings import settings
 from blossomtune_gradio.database import SessionLocal, Request, Config
+from blossomtune_gradio.auth_keys import AuthKeyGenerator, rebuild_authorized_keys_csv
+from blossomtune_gradio.blossomfile import create_blossomfile
 
 
 def generate_participant_id(length=6):
@@ -24,7 +28,6 @@ def check_participant_status(pid_to_check: str, email: str, activation_code: str
     """
     Handles a participant's request to join, activate, or check status using SQLAlchemy.
     Returns a tuple: (is_approved: bool, message: str, data: any | None)
-    The 'is_approved' boolean is True ONLY when the participant's final status is 'approved'.
     """
     with SessionLocal() as db:
         query = db.query(Request).filter(
@@ -40,51 +43,46 @@ def check_participant_status(pid_to_check: str, email: str, activation_code: str
         )
         num_partitions = num_partitions_config.value if num_partitions_config else "10"
 
-        # Case 1: New user registration
-        if request is None:
-            if activation_code:
-                return (False, settings.get_text("activation_invalid_md"), None)
-            if not util.validate_email(email):
-                return (False, settings.get_text("invalid_email_md"), None)
-
-            approved_count = (
-                db.query(Request).filter(Request.status == "approved").count()
-            )
-            if approved_count >= cfg.MAX_NUM_NODES:
-                return (False, settings.get_text("federation_full_md"), None)
-
-            participant_id = generate_participant_id()
-            new_activation_code = generate_activation_code()
-            mail_sent, message = mail.send_activation_email(email, new_activation_code)
-
-            if mail_sent:
-                new_request = Request(
-                    participant_id=participant_id,
-                    hf_handle=pid_to_check,
-                    email=email,
-                    activation_code=new_activation_code,
+        # Case 1 & 2 are for users not yet approved
+        if not request or not request.is_activated or not activation_code:
+            if request is None:
+                if activation_code:
+                    return (False, settings.get_text("activation_invalid_md"), None)
+                if not util.validate_email(email):
+                    return (False, settings.get_text("invalid_email_md"), None)
+                approved_count = (
+                    db.query(Request).filter(Request.status == "approved").count()
+                )
+                if approved_count >= cfg.MAX_NUM_NODES:
+                    return (False, settings.get_text("federation_full_md"), None)
+                participant_id = generate_participant_id()
+                new_activation_code = generate_activation_code()
+                mail_sent, message = mail.send_activation_email(
+                    email, new_activation_code
                 )
-                db.add(new_request)
-                db.commit()
-                # A successful registration step, but not yet approved for federation.
-                return (False, settings.get_text("registration_submitted_md"), None)
-            else:
-                return (False, message, None)
-
-        # Case 2: User is activating their account
-        if not request.is_activated:
-            if activation_code == request.activation_code:
-                request.is_activated = 1
-                db.commit()
-                # A successful activation step, but not yet approved.
-                return (False, settings.get_text("activation_successful_md"), None)
-            else:
-                return (False, settings.get_text("activation_invalid_md"), None)
-
-        # At this point, user is activated.
-        # They must provide the activation code to check their final status.
-        if not activation_code:
-            return (False, settings.get_text("missing_activation_code_md"), None)
+                if mail_sent:
+                    new_request = Request(
+                        participant_id=participant_id,
+                        hf_handle=pid_to_check,
+                        email=email,
+                        activation_code=new_activation_code,
+                    )
+                    db.add(new_request)
+                    db.commit()
+                    return (False, settings.get_text("registration_submitted_md"), None)
+                else:
+                    return (False, message, None)
+
+            if not request.is_activated:
+                if activation_code == request.activation_code:
+                    request.is_activated = 1
+                    db.commit()
+                    return (False, settings.get_text("activation_successful_md"), None)
+                else:
+                    return (False, settings.get_text("activation_invalid_md"), None)
+
+            if not activation_code:
+                return (False, settings.get_text("missing_activation_code_md"), None)
 
         # Case 3: Activated user is checking their final status
         if request.status == "approved":
@@ -93,17 +91,37 @@ def check_participant_status(pid_to_check: str, email: str, activation_code: str
                 if not cfg.SPACE_ID
                 else f"{cfg.SPACE_ID.split('/')[1]}-{cfg.SPACE_ID.split('/')[0]}.hf.space"
             )
-            superlink_hostname = cfg.SUPERLINK_HOST or hostname
+            superlink_address = f"{cfg.SUPERLINK_HOST or hostname}:{cfg.SUPERLINK_PORT}"
+
+            # Blossomfile Generation
+            blossomfile_tempdir = tempfile.mkdtemp()  # TODO: remove tempdirs
+            try:
+                blossomfile_path = create_blossomfile(
+                    participant_id=request.participant_id,
+                    output_dir=blossomfile_tempdir,
+                    ca_cert_path=cfg.BLOSSOMTUNE_TLS_CA_CERTFILE,
+                    auth_key_path=os.path.join(
+                        cfg.AUTH_KEYS_DIR, f"{request.participant_id}.key"
+                    ),
+                    auth_pub_path=os.path.join(
+                        cfg.AUTH_KEYS_DIR, f"{request.participant_id}.pub"
+                    ),
+                    superlink_address=superlink_address,
+                    partition_id=request.partition_id,
+                    num_partitions=int(num_partitions),
+                )
+            except FileNotFoundError:
+                return (False, "An error occurred.", None)
 
             connection_string = settings.get_text(
                 "status_approved_md",
                 participant_id=request.participant_id,
                 partition_id=request.partition_id,
-                superlink_hostname=superlink_hostname,
+                superlink_hostname=superlink_address.split(":")[0],
+                superlink_port=superlink_address.split(":")[1],
                 num_partitions=num_partitions,
             )
-            # The user is fully approved. Return success and the cert path.
-            return (True, connection_string, cfg.BLOSSOMTUNE_TLS_CA_CERTFILE)
+            return (True, connection_string, blossomfile_path)
         elif request.status == "pending":
             return (False, settings.get_text("status_pending_md"), None)
         else:  # Denied
@@ -131,7 +149,6 @@ def manage_request(participant_id: str, partition_id: str, action: str):
         if action == "approve":
             if not partition_id or not partition_id.isdigit():
                 return False, "Please provide a valid integer for the Partition ID."
-
             p_id_int = int(partition_id)
             if not request.is_activated:
                 return (
@@ -144,7 +161,6 @@ def manage_request(participant_id: str, partition_id: str, action: str):
                 .filter(Request.partition_id == p_id_int, Request.status == "approved")
                 .first()
             )
-
             if existing_participant:
                 return (
                     False,
@@ -155,18 +171,48 @@ def manage_request(participant_id: str, partition_id: str, action: str):
 
             request.status = "approved"
             request.partition_id = p_id_int
+
+            # Generate and Store Auth Keys
+            key_generator = AuthKeyGenerator(key_dir=cfg.AUTH_KEYS_DIR)
+            _, _, public_key_pem = key_generator.generate_participant_keys(
+                participant_id
+            )
+            request.public_key_pem = public_key_pem
             db.commit()
+
+            # Rebuild Authorized Keys CSV
+            approved_participants = (
+                db.query(Request.participant_id, Request.public_key_pem)
+                .filter(
+                    Request.status == "approved", Request.public_key_pem.isnot(None)
+                )
+                .all()
+            )
+            rebuild_authorized_keys_csv(cfg.AUTH_KEYS_DIR, approved_participants)
+
             return (
                 True,
-                f"Participant {participant_id} is allowed to join the federation.",
+                f"Participant {participant_id} approved. Keys generated and registry updated.",
             )
         else:  # Deny
             request.status = "denied"
             request.partition_id = None
+            request.public_key_pem = None
             db.commit()
+
+            # --- Rebuild CSV after denial to revoke access ---
+            approved_participants = (
+                db.query(Request.participant_id, Request.public_key_pem)
+                .filter(
+                    Request.status == "approved", Request.public_key_pem.isnot(None)
+                )
+                .all()
+            )
+            rebuild_authorized_keys_csv(cfg.AUTH_KEYS_DIR, approved_participants)
+
             return (
                 True,
-                f"Participant {participant_id} is not allowed to join the federation.",
+                f"Participant {participant_id} denied. Their access has been revoked.",
             )
 
 

blossomtune_gradio/processing.py

@@ -54,7 +54,9 @@ def start_superlink():
         cfg.BLOSSOMTUNE_TLS_CERTFILE,
         "--ssl-keyfile",
         cfg.BLOSSOMTUNE_TLS_KEYFILE,
-    ]  # Placeholder
+        "--auth-list-public-keys",
+        cfg.AUTH_KEYS_CSV_PATH,
+    ]
     threading.Thread(
         target=run_process, args=(command, "superlink"), daemon=True
     ).start()

blossomtune_gradio/settings/blossomtune.schema.json

@@ -100,7 +100,7 @@
         "status_pending_md",
         "status_denied_md",
         "participant_not_activated_warning_md",
-        "partition_in_use_warning",
+        "partition_in_use_warning_md",
         "auth_status_logged_in_owner_md",
         "auth_status_logged_in_user_md",
         "auth_status_not_logged_in_md",

blossomtune_gradio/settings/blossomtune.yaml

@@ -66,7 +66,7 @@ ui:
   participant_not_activated_warning_md: |
     This participant has not activated their email yet. Approval is not allowed.
     
-  partition_in_use_warning: |
+  partition_in_use_warning_md: |
     Partition ID {{ partition_id }} is already assigned. Please choose a different one.
 
   # --- Authentication Status ---

docker_entrypoint.sh

@@ -1,5 +1,4 @@
 #!/bin/bash
-
 set -e
 
 echo "Env vars:"
@@ -13,7 +12,8 @@ elif [ "${1}" = "superlink" ]; then
     exec flower-superlink \
         --ssl-ca-certfile /data/certs/ca.crt \
         --ssl-certfile /data/certs/server.pem \
-        --ssl-keyfile /data/certs/server.key
+        --ssl-keyfile /data/certs/server.key \
+        --auth-list-public-keys /data/keys/authorized_supernodes.csv
 else
     exec "$@"
 fi

pyproject.toml

@@ -18,6 +18,9 @@ dependencies = [
     "mlx[cpu]>=0.29.2",
     "alembic>=1.16.5",
     "sqlalchemy>=2.0.43",
+    "cryptography>=44.0.3",
+    "dnspython>=2.8.0",
+    "mlx-lm>=0.28.2",
 ]
 
 [tool.uv.sources]
@@ -71,8 +74,6 @@ convention = "google"  # Accepts: "google", "numpy", or "pep257".
 
 [dependency-groups]
 dev = [
-    "cryptography>=44.0.3",
-    "dnspython>=2.8.0",
     "pytest>=8.4.1",
     "pytest-mock>=3.15.1",
 ]

requirements.txt

@@ -0,0 +1,16 @@
+apscheduler>=3.11.0
+flwr[simulation]>=1.21.0
+flwr-datasets>=0.5.0
+gradio[oauth]>=5.44.1
+torch>=2.8.0
+transformers>=4.56.1
+scikit-learn>=1.7.1
+evaluate>=0.4.5
+markupsafe==2.1.3
+jinja2>=3.1.6
+# mlx[cpu]>=0.29.2
+alembic>=1.16.5
+sqlalchemy>=2.0.43
+cryptography>=44.0.3
+dnspython>=2.8.0
+# mlx-lm>=0.28.2

tests/test_auth_keys.py

@@ -1,5 +1,4 @@
 import os
-import csv
 import stat
 import pytest
 from cryptography.hazmat.primitives import serialization
@@ -25,109 +24,117 @@ class TestAuthKeyGenerator:
         AuthKeyGenerator(key_dir=str(key_dir))
         assert os.path.exists(key_dir)
 
-    def test_generate_participant_keys_creates_files_and_returns_data(
+    def test_generate_participant_keys_creates_files_and_returns_openssh_with_comment(
         self, key_generator
     ):
         """
-        Verify that the main method generates all expected files and returns
-        the correct data tuple.
+        Verify the main method generates files and returns the public key
+        in the correct OpenSSH format including a comment.
         """
         participant_id = "participant_01"
-        priv_path, pub_path, pub_pem = key_generator.generate_participant_keys(
+        priv_path, pub_path, pub_ssh_string = key_generator.generate_participant_keys(
             participant_id
         )
 
-        # 1. Check if files exist
+        # 1. Check file existence and permissions
         assert os.path.exists(priv_path)
         assert os.path.exists(pub_path)
-        assert priv_path == os.path.join(key_generator.key_dir, f"{participant_id}.key")
-        assert pub_path == os.path.join(key_generator.key_dir, f"{participant_id}.pub")
-
-        # 2. Check private key file permissions (security check)
-        # In non-Windows environments, check for 600 permissions.
         if os.name != "nt":
             file_mode = stat.S_IMODE(os.stat(priv_path).st_mode)
             assert file_mode == 0o600
 
-        # 3. Verify key formats and consistency
-        with open(priv_path, "rb") as f:
-            private_key = serialization.load_pem_private_key(f.read(), password=None)
-        with open(pub_path, "rb") as f:
-            public_key = serialization.load_pem_public_key(f.read())
-            _ = f.read()  # Read again to get bytes
-
-        assert isinstance(private_key, ec.EllipticCurvePrivateKey)
-        assert isinstance(public_key, ec.EllipticCurvePublicKey)
+        # 2. Verify that the returned string has three parts (type, key, comment)
+        assert pub_ssh_string.startswith("ecdsa-sha2-nistp384")
+        assert pub_ssh_string.endswith(participant_id)
+        assert len(pub_ssh_string.split(" ")) == 3
 
-        # Check that the returned PEM string matches the public key
-        generated_public_key = private_key.public_key()
-        pem_from_private = generated_public_key.public_bytes(
-            encoding=serialization.Encoding.PEM,
-            format=serialization.PublicFormat.SubjectPublicKeyInfo,
-        ).decode("utf-8")
-        assert pub_pem == pem_from_private
+        # 3. Verify that the public key file can be loaded as an SSH key
+        with open(pub_path, "rb") as f:
+            public_key_from_file = serialization.load_ssh_public_key(f.read())
+        assert isinstance(public_key_from_file, ec.EllipticCurvePublicKey)
 
 
-class TestRebuildCSV:
+class TestRebuildAuthorizedKeysFile:
     """Test suite for the rebuild_authorized_keys_csv function."""
 
-    def test_rebuild_csv_creates_file_with_header_for_empty_list(self, tmp_path):
-        """Verify a CSV with only a header is created for an empty participant list."""
-        key_dir = tmp_path / "csv_test"
+    def test_rebuild_creates_file_with_only_newline_for_empty_list(self, tmp_path):
+        """Verify an empty participant list results in a file with just a newline."""
+        key_dir = tmp_path / "keys_test"
         os.makedirs(key_dir)
-        csv_path = os.path.join(key_dir, "authorized_supernodes.csv")
+        rebuild_authorized_keys_csv(str(key_dir), [])
 
-        rebuild_authorized_keys_csv(key_dir, [])
-
-        assert os.path.exists(csv_path)
+        csv_path = os.path.join(key_dir, "authorized_supernodes.csv")
         with open(csv_path, "r") as f:
-            reader = csv.reader(f)
-            header = next(reader)
-            assert header == ["participant_id", "public_key_pem"]
-            # Check that there are no more rows
-            with pytest.raises(StopIteration):
-                next(reader)
-
-    def test_rebuild_csv_writes_correct_data(self, tmp_path):
-        """Verify the CSV is created with the correct participant data."""
-        key_dir = tmp_path / "csv_test"
+            content = f.read()
+        assert content == "\n"
+
+    def test_rebuild_writes_correct_single_line_format(self, tmp_path):
+        """Verify the file is created in the single-line, comma-separated format."""
+        key_dir = tmp_path / "keys_test"
         os.makedirs(key_dir)
-        csv_path = os.path.join(key_dir, "authorized_supernodes.csv")
 
         participants = [
-            ("p1", "---BEGIN PUBLIC KEY---...p1...---END PUBLIC KEY---"),
-            ("p2", "---BEGIN PUBLIC KEY---...p2...---END PUBLIC KEY---"),
+            ("p1", "ecdsa-sha2-nistp384 AAAA...key1 p1"),
+            ("p2", "ecdsa-sha2-nistp384 AAAA...key2 p2"),
         ]
-        rebuild_authorized_keys_csv(key_dir, participants)
+        rebuild_authorized_keys_csv(str(key_dir), participants)
 
+        csv_path = os.path.join(key_dir, "authorized_supernodes.csv")
         with open(csv_path, "r") as f:
-            reader = csv.reader(f)
-            header = next(reader)
-            row1 = next(reader)
-            row2 = next(reader)
-            assert header == ["participant_id", "public_key_pem"]
-            assert row1 == list(participants[0])
-            assert row2 == list(participants[1])
-
-    def test_rebuild_csv_overwrites_existing_file(self, tmp_path):
-        """Verify that an existing CSV file is correctly overwritten."""
-        key_dir = tmp_path / "csv_test"
+            content = f.read().strip()
+
+        expected_content = (
+            "ecdsa-sha2-nistp384 AAAA...key1 p1,ecdsa-sha2-nistp384 AAAA...key2 p2"
+        )
+        assert content == expected_content
+
+    def test_rebuild_overwrites_existing_file(self, tmp_path):
+        """Verify that an existing file is correctly overwritten."""
+        key_dir = tmp_path / "keys_test"
         os.makedirs(key_dir)
+
+        # Use dummy data that matches the expected OpenSSH format
+        initial_participants = [("old_p1", "ecdsa-sha2-nistp384 old_key_1 old_p1")]
+        rebuild_authorized_keys_csv(str(key_dir), initial_participants)
+
+        new_participants = [
+            ("new_p1", "ecdsa-sha2-nistp384 new_key_1 new_p1"),
+            ("new_p2", "ecdsa-sha2-nistp384 new_key_2 new_p2"),
+        ]
+        rebuild_authorized_keys_csv(str(key_dir), new_participants)
+
         csv_path = os.path.join(key_dir, "authorized_supernodes.csv")
+        with open(csv_path, "r") as f:
+            content = f.read().strip()
 
-        # First run with initial data
-        initial_participants = [("old_p1", "old_key_1")]
-        rebuild_authorized_keys_csv(key_dir, initial_participants)
+        expected_content = (
+            "ecdsa-sha2-nistp384 new_key_1 new_p1,ecdsa-sha2-nistp384 new_key_2 new_p2"
+        )
+        assert content == expected_content
 
-        # Second run with new data
-        new_participants = [("new_p1", "new_key_1"), ("new_p2", "new_key_2")]
-        rebuild_authorized_keys_csv(key_dir, new_participants)
+    def test_rebuild_sanitizes_pem_keys_to_ssh_format(self, tmp_path):
+        """
+        Tests the self-healing capability of the rebuild function to convert
+        old PEM keys from the database into the correct OpenSSH format.
+        """
+        key_dir = tmp_path / "keys_test"
+        os.makedirs(key_dir)
 
+        # Generate a real key pair to get a valid PEM string
+        private_key = ec.generate_private_key(ec.SECP384R1())
+        public_key = private_key.public_key()
+        pem_key = public_key.public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        ).decode("utf-8")
+
+        participants = [("p1_pem", pem_key)]
+        rebuild_authorized_keys_csv(str(key_dir), participants)
+
+        csv_path = os.path.join(key_dir, "authorized_supernodes.csv")
         with open(csv_path, "r") as f:
-            reader = csv.reader(f)
-            _ = next(reader)
-            rows = list(reader)
+            content = f.read().strip()
 
-        assert len(rows) == 2
-        assert rows[0] == list(new_participants[0])
-        assert rows[1] == list(new_participants[1])
+        # Verify the output is now in the correct OpenSSH format with the comment
+        assert content.startswith("ecdsa-sha2-nistp384")
+        assert content.endswith("p1_pem")

tests/test_blossomfile.py

@@ -50,7 +50,7 @@ def test_create_blossomfile_success(tmp_path, dummy_credential_files):
 
     # 1. Verify the file was created at the correct path
     assert os.path.exists(blossomfile_path)
-    assert blossomfile_path == str(output_dir / f"{participant_id}.blossomfile")
+    assert blossomfile_path == str(output_dir / f"{participant_id}-blossomfile.zip")
 
     # 2. Verify the contents of the zip archive
     with zipfile.ZipFile(blossomfile_path, "r") as zf:

tests/test_federation.py

@@ -34,7 +34,7 @@ class TestCheckParticipantStatus:
         """Verify successful registration for a new user."""
         mock_mail.return_value = (True, "")
         approved, message, download = fed.check_participant_status(
-            "new_user", "new@example.com", ""
+            "new_user", "hello@ethicalabs.ai", ""
         )
         assert approved is False
         assert download is None
@@ -43,7 +43,7 @@ class TestCheckParticipantStatus:
         # Verify the user was added to the database
         request = db_session.query(Request).filter_by(hf_handle="new_user").first()
         assert request is not None
-        assert request.email == "new@example.com"
+        assert request.email == "hello@ethicalabs.ai"
 
     def test_new_user_invalid_email(self, db_session, mock_settings):
         """Verify registration fails with an invalid email."""
@@ -108,7 +108,7 @@ class TestCheckParticipantStatus:
         assert download is None
         assert message == "mock_activation_invalid_md"
 
-    def test_status_check_approved(self, db_session, mock_settings):
+    def test_status_check_approved_unmanaged(self, db_session, mock_settings):
         """Verify the status check for an approved user."""
         approved_user = Request(
             participant_id="PID456",
@@ -125,9 +125,9 @@ class TestCheckParticipantStatus:
         approved, message, download = fed.check_participant_status(
             "approved_user", "approved@example.com", "GHIJKL"
         )
-        assert approved is True
-        assert download is not None
-        assert "mock_status_approved_md" in message
+        assert approved is False
+        assert download is None  # fed.manage_request is not used. download is None.
+        assert "An error occurred" in message
 
 
 class TestManageRequest:
@@ -147,7 +147,10 @@ class TestManageRequest:
 
         success, message = fed.manage_request("PENDING1", "10", "approve")
         assert success is True
-        assert "is allowed to join" in message
+        assert (
+            "Participant PENDING1 approved. Keys generated and registry updated."
+            in message
+        )
 
         # Verify status in DB
         updated_user = (
@@ -182,7 +185,7 @@ class TestManageRequest:
         db_session.commit()
         success, message = fed.manage_request("PENDING3", "", "deny")
         assert success is True
-        assert "is not allowed to join" in message
+        assert "Participant PENDING3 denied. Their access has been revoked." in message
 
         # Verify status in DB
         updated_user = (