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}

docker-compose.yml

@@ -239,7 +239,7 @@ services:
       - "5434:5432"
     networks:
       - fmtm-net
-    restart: "on-failure:3"
+    restart: "unless-stopped"
     healthcheck:
       test: pg_isready -U ${CENTRAL_DB_USER:-odk} -d ${CENTRAL_DB_NAME:-odk}
       start_period: 5s

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
@@ -43,7 +40,7 @@ def get_odk_project(odk_central: project_schemas.ODKCentral = None):
     if odk_central:
         url = odk_central.odk_central_url
         user = odk_central.odk_central_user
-        pw = odk_central.odk_central_password
+        pw = odk_central.odk_central_password.get_secret_value()
     else:
         log.debug("ODKCentral connection variables not set in function")
         log.debug("Attempting extraction from environment variables")
@@ -68,7 +65,7 @@ def get_odk_form(odk_central: project_schemas.ODKCentral = None):
     if odk_central:
         url = odk_central.odk_central_url
         user = odk_central.odk_central_user
-        pw = odk_central.odk_central_password
+        pw = odk_central.odk_central_password.get_secret_value()
 
     else:
         log.debug("ODKCentral connection variables not set in function")
@@ -94,7 +91,7 @@ def get_odk_app_user(odk_central: project_schemas.ODKCentral = None):
     if odk_central:
         url = odk_central.odk_central_url
         user = odk_central.odk_central_user
-        pw = odk_central.odk_central_password
+        pw = odk_central.odk_central_password.get_secret_value()
     else:
         log.debug("ODKCentral connection variables not set in function")
         log.debug("Attempting extraction from environment variables")
@@ -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
 ):
@@ -226,7 +193,7 @@ def upload_xform_media(
             status_code=500, detail={"message": "Connection failed to odk central"}
         ) from e
 
-    result = xform.uploadMedia(project_id, title, filespec)
+    xform.uploadMedia(project_id, title, filespec)
     result = xform.publishForm(project_id, title)
     return result
 
@@ -268,9 +235,7 @@ def create_odk_xform(
     # This modifies an existing published XForm to be in draft mode.
     # An XForm must be in draft mode to upload an attachment.
     if upload_media:
-        result = xform.uploadMedia(
-            project_id, title, data, convert_to_draft_when_publishing
-        )
+        xform.uploadMedia(project_id, title, data, convert_to_draft_when_publishing)
 
     result = xform.publishForm(project_id, title)
     return result
@@ -537,38 +502,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()

src/backend/app/organisations/organisation_schemas.py

@@ -21,9 +21,10 @@
 from typing import Optional
 
 from fastapi import Form
-from pydantic import BaseModel, Field, HttpUrl, computed_field
+from pydantic import BaseModel, Field, HttpUrl, SecretStr, computed_field
 from pydantic.functional_validators import field_validator
 
+from app.config import decrypt_value, encrypt_value
 from app.models.enums import OrganisationType
 
 # class OrganisationBase(BaseModel):
@@ -37,7 +38,16 @@ class OrganisationIn(BaseModel):
     description: Optional[str] = Field(
         Form(None, description="Organisation description")
     )
-    url: Optional[HttpUrl] = Field(Form(None, description="Organisation website URL"))
+    url: Optional[HttpUrl] = Field(Form(None, description=("Organisation website URL")))
+    odk_central_url: Optional[str] = Field(
+        Form(None, description="Organisation default ODK URL")
+    )
+    odk_central_user: Optional[str] = Field(
+        Form(None, description="Organisation default ODK User")
+    )
+    odk_central_password: Optional[SecretStr] = Field(
+        Form(None, description="Organisation default ODK Password")
+    )
 
     @field_validator("url", mode="after")
     @classmethod
@@ -54,12 +64,21 @@ def convert_url_to_str(cls, value: HttpUrl) -> str:
     @property
     def slug(self) -> str:
         """Sanitise the organisation name for use in a URL."""
-        if self.name:
-            # Remove special characters and replace spaces with hyphens
-            slug = sub(r"[^\w\s-]", "", self.name).strip().lower().replace(" ", "-")
-            # Remove consecutive hyphens
-            slug = sub(r"[-\s]+", "-", slug)
-            return slug
+        if not self.name:
+            return ""
+        # Remove special characters and replace spaces with hyphens
+        slug = sub(r"[^\w\s-]", "", self.name).strip().lower().replace(" ", "-")
+        # Remove consecutive hyphens
+        slug = sub(r"[-\s]+", "-", slug)
+        return slug
+
+    @field_validator("odk_central_password", mode="before")
+    @classmethod
+    def encrypt_odk_password(cls, value: str) -> Optional[SecretStr]:
+        """Encrypt the ODK Central password before db insertion."""
+        if not value:
+            return None
+        return SecretStr(encrypt_value(value))
 
 
 class OrganisationEdit(OrganisationIn):
@@ -79,3 +98,22 @@ class OrganisationOut(BaseModel):
     slug: Optional[str]
     url: Optional[str]
     type: OrganisationType
+    odk_central_url: Optional[str] = None
+
+
+class OrganisationOutWithCreds(BaseModel):
+    """Organisation plus ODK Central credentials.
+
+    Note: the password is obsfucated as SecretStr.
+    """
+
+    odk_central_user: Optional[str] = None
+    odk_central_password: Optional[SecretStr] = None
+
+    def model_post_init(self, ctx):
+        """Run logic after model object instantiated."""
+        # Decrypt odk central password from database
+        if self.odk_central_password:
+            self.odk_central_password = SecretStr(
+                decrypt_value(self.odk_central_password.get_secret_value())
+            )