Dynamic qrcode generation via frontend

.env.example

@@ -7,6 +7,7 @@ ODK_CENTRAL_PASSWD=${ODK_CENTRAL_PASSWD:-"testuserpassword"}
 DEBUG=${DEBUG:-False}
 LOG_LEVEL=${LOG_LEVEL:-INFO}
 EXTRA_CORS_ORIGINS=${EXTRA_CORS_ORIGINS}
+ENCRYPTION_KEY=${ENCRYPTION_KEY:-"pIxxYIXe4oAVHI36lTveyc97FKK2O_l2VHeiuqU-K_4="}
 FMTM_DOMAIN=${FMTM_DOMAIN:-"fmtm.localhost"}
 FMTM_DEV_PORT=${FMTM_DEV_PORT:-7050}
 CERT_EMAIL=${CERT_EMAIL}

scripts/qrcode_to_odktoken.py

@@ -0,0 +1,148 @@
+"""Convert a QR Code image in Postgres to a Fernet encrypted odk_token URL."""
+
+from pathlib import Path
+from io import BytesIO
+
+import argparse
+import base64
+import zlib
+import json
+from segno import make as make_qr
+# apt install libzbar-dev 
+from pyzbar.pyzbar import decode as decode_qr
+# pip install pillow
+from PIL import Image
+from sqlalchemy import ForeignKey, Column, Integer, String, LargeBinary
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm.attributes import InstrumentedAttribute
+
+from dotenv import load_dotenv
+
+load_dotenv(Path(__file__).parent.parent / ".env.example")
+
+from app.config import encrypt_value, decrypt_value  # noqa: E402
+from app.db.database import Base, get_db
+from app.db.db_models import DbProject, DbTask
+
+
+class DbQrCode(Base):
+    """QR Code."""
+
+    __tablename__ = "qr_code"
+
+    id = Column(Integer, primary_key=True)
+    filename = Column(String)
+    image = Column(LargeBinary)
+
+
+class TaskPlusQR(DbTask):
+    """Task plus QR code foreign key."""
+    qr_code_id = Column(Integer, ForeignKey('qr_code.id'), index=True)
+    qr_code = relationship(
+        DbQrCode, cascade="all", single_parent=True
+    )
+    if not isinstance(DbTask.odk_token, InstrumentedAttribute):
+        odk_token = Column(String, nullable=True)
+
+
+def odktoken_to_qr():
+    """Extract odk_token field from db and convert to QR codes."""
+
+    db = next(get_db())
+    projects = db.query(DbProject).all()
+
+    for project in projects:
+        project_name = project.project_name_prefix
+        tasks = project.tasks
+
+        for task in tasks:
+            odk_token = task.odk_token
+            if not odk_token:
+                continue
+
+            decrypted_odk_token = decrypt_value(odk_token)
+            qr_code_setting = {
+                "general": {
+                    "server_url": decrypted_odk_token,
+                    "form_update_mode": "match_exactly",
+                    "basemap_source": "osm",
+                    "autosend": "wifi_and_cellular",
+                    "metadata_username": "svcfmtm",
+                },
+                "project": {"name": f"{project_name}"},
+                "admin": {},
+            }
+
+            # Base64/zlib encoded
+            qrcode_data = base64.b64encode(
+                zlib.compress(json.dumps(qr_code_setting).encode("utf-8"))
+            )
+            qrcode = make_qr(qrcode_data, micro=False)
+            buffer = BytesIO()
+            qrcode.save(buffer, kind="png", scale=5)
+            qrcode_binary = buffer.getvalue()
+            qrdb = DbQrCode(image=qrcode_binary)
+            db.add(qrdb)
+            print(f"Added qrcode for task {task.id} to db")
+    db.commit()
+
+
+def qr_to_odktoken():
+    """Extract QR codes from db and convert to odk_token field."""
+
+    db = next(get_db())
+    tasks = db.query(TaskPlusQR).all()
+
+    for task in tasks:
+        if task.qr_code:
+            qr_img = Image.open(BytesIO(task.qr_code.image))
+            qr_data = decode_qr(qr_img)[0].data
+
+            # Base64/zlib decoded
+            decoded_qr = zlib.decompress(base64.b64decode(qr_data))
+            odk_token = json.loads(decoded_qr.decode("utf-8")).get("general", {}).get("server_url")
+
+            task.odk_token = encrypt_value(odk_token)
+            print(f"Added odk token for task {task.id}")
+    db.commit()
+
+
+def encrypt_odk_creds():
+    """Encrypt project odk password in the db."""
+
+    db = next(get_db())
+    projects = db.query(DbProject).all()
+
+    for project in projects:
+        project.odk_central_password = encrypt_value(project.odk_central_password)
+        print(f"Encrypted odk password for project {project.id}")
+    db.commit()
+
+
+def decrypt_odk_creds():
+    """Decrypt project odk password in the db."""
+
+    db = next(get_db())
+    projects = db.query(DbProject).all()
+
+    for project in projects:
+        project.odk_central_password = decrypt_value(project.odk_central_password)
+        print(f"Encrypted odk password for project {project.id}")
+    db.commit()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Apply or revert changes to QR codes and odk tokens.")
+    parser.add_argument("--apply", action="store_true", help="Apply changes (convert QR codes to odk tokens).")
+    parser.add_argument("--revert", action="store_true", help="Revert changes (convert odk tokens to QR codes).")
+
+    args = parser.parse_args()
+
+    if args.apply:
+        qr_to_odktoken()
+        encrypt_odk_creds()
+    elif args.revert:
+        odktoken_to_qr()
+        decrypt_odk_creds()
+    else:
+        print("Please provide either --apply or --revert flag.")

src/backend/app/central/central_crud.py

@@ -17,10 +17,7 @@
 #
 """Logic for interaction with ODK Central & data."""
 
-import base64
-import json
 import os
-import zlib
 from xml.etree import ElementTree
 
 # import osm_fieldwork
@@ -161,36 +158,6 @@ async def delete_odk_project(
         return "Could not delete project from central odk"
 
 
-def create_odk_app_user(
-    project_id: int, name: str, odk_credentials: project_schemas.ODKCentral = None
-):
-    """Create an app user specific to a project on ODK Central.
-
-    If odk credentials of the project are provided, use them to create an app user.
-    """
-    if odk_credentials:
-        url = odk_credentials.odk_central_url
-        user = odk_credentials.odk_central_user
-        pw = odk_credentials.odk_central_password
-
-    else:
-        log.debug("ODKCentral connection variables not set in function")
-        log.debug("Attempting extraction from environment variables")
-        url = settings.ODK_CENTRAL_URL
-        user = settings.ODK_CENTRAL_USER
-        pw = settings.ODK_CENTRAL_PASSWD
-
-    odk_app_user = OdkAppUser(url, user, pw)
-
-    log.debug(
-        "ODKCentral: attempting user creation: name: " f"{name} | project: {project_id}"
-    )
-    result = odk_app_user.create(project_id, name)
-
-    log.debug(f"ODKCentral response: {result.json()}")
-    return result
-
-
 def delete_odk_app_user(
     project_id: int, name: str, odk_central: project_schemas.ODKCentral = None
 ):
@@ -537,38 +504,6 @@ def generate_updated_xform(
     return outfile
 
 
-async def encode_qrcode_json(
-    project_id: int, token: str, name: str, odk_central_url: str = None
-):
-    """Assemble the ODK Collect JSON and base64 encode.
-
-    The base64 encoded string is used to generate a QR code later.
-    """
-    if not odk_central_url:
-        log.debug("ODKCentral connection variables not set in function")
-        log.debug("Attempting extraction from environment variables")
-        odk_central_url = settings.ODK_CENTRAL_URL
-
-    # QR code text json in the format acceptable by odk collect
-    qr_code_setting = {
-        "general": {
-            "server_url": f"{odk_central_url}/v1/key/{token}/projects/{project_id}",
-            "form_update_mode": "match_exactly",
-            "basemap_source": "osm",
-            "autosend": "wifi_and_cellular",
-            "metadata_username": "svcfmtm",
-        },
-        "project": {"name": f"{name}"},
-        "admin": {},
-    }
-
-    # Base64 encoded
-    qr_data = base64.b64encode(
-        zlib.compress(json.dumps(qr_code_setting).encode("utf-8"))
-    )
-    return qr_data
-
-
 def upload_media(
     project_id: int,
     xform_id: str,

src/backend/app/central/central_schemas.py

@@ -32,7 +32,6 @@ class Central(CentralBase):
     """ODK Central return, with extras."""
 
     geometry_geojson: str
-    # qr_code_binary: bytes
 
 
 class CentralOut(CentralBase):

src/backend/app/config.py

@@ -17,9 +17,11 @@
 #
 """Config file for Pydantic and FastAPI, using environment variables."""
 
+import base64
 from functools import lru_cache
 from typing import Any, Optional, Union
 
+from cryptography.fernet import Fernet
 from pydantic import PostgresDsn, ValidationInfo, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -30,6 +32,7 @@ class Settings(BaseSettings):
     APP_NAME: str = "FMTM"
     DEBUG: bool = False
     LOG_LEVEL: str = "INFO"
+    ENCRYPTION_KEY: str = ""
 
     FMTM_DOMAIN: str
     FMTM_DEV_PORT: Optional[str] = "7050"
@@ -161,4 +164,25 @@ def get_settings():
     return _settings
 
 
+@lru_cache
+def get_cipher_suite():
+    """Cache cypher suite."""
+    return Fernet(settings.ENCRYPTION_KEY)
+
+
+def encrypt_value(password: str) -> str:
+    """Encrypt value before going to the DB."""
+    cipher_suite = get_cipher_suite()
+    encrypted_password = cipher_suite.encrypt(password.encode("utf-8"))
+    return base64.b64encode(encrypted_password).decode("utf-8")
+
+
+def decrypt_value(db_password: str) -> str:
+    """Decrypt the database value."""
+    cipher_suite = get_cipher_suite()
+    encrypted_password = base64.b64decode(db_password.encode("utf-8"))
+    decrypted_password = cipher_suite.decrypt(encrypted_password)
+    return decrypted_password.decode("utf-8")
+
+
 settings = get_settings()

src/backend/app/db/db_models.py

@@ -149,6 +149,11 @@ class DbOrganisation(Base):
     type = Column(Enum(OrganisationType), default=OrganisationType.FREE, nullable=False)
     approved = Column(Boolean, default=False)
 
+    ## Odk central server
+    odk_central_url = Column(String)
+    odk_central_user = Column(String)
+    odk_central_password = Column(String)
+
     managers = relationship(
         DbUser,
         secondary=organisation_managers,
@@ -345,16 +350,6 @@ class DbTaskHistory(Base):
     )
 
 
-class DbQrCode(Base):
-    """QR Code."""
-
-    __tablename__ = "qr_code"
-
-    id = Column(Integer, primary_key=True)
-    filename = Column(String)
-    image = Column(LargeBinary)
-
-
 class DbTask(Base):
     """Describes an individual mapping Task."""
 
@@ -380,13 +375,9 @@ class DbTask(Base):
     validated_by = Column(
         BigInteger, ForeignKey("users.id", name="fk_users_validator"), index=True
     )
+    odk_token = Column(String, nullable=True)
 
     # Mapped objects
-    qr_code_id = Column(Integer, ForeignKey("qr_code.id"), index=True)
-    qr_code = relationship(
-        DbQrCode, cascade="all, delete, delete-orphan", single_parent=True
-    )
-
     task_history = relationship(
         DbTaskHistory, cascade="all", order_by=desc(DbTaskHistory.action_date)
     )

src/backend/app/organisations/organisation_crud.py

@@ -106,7 +106,7 @@ async def create_organisation(
 
     try:
         # Create new organisation without logo set
-        db_organisation = db_models.DbOrganisation(**org_model.dict())
+        db_organisation = db_models.DbOrganisation(**org_model.model_dump())
 
         db.add(db_organisation)
         db.commit()