Default secrets

SSG-API-Testing-Application-v2/app/Dockerfile

@@ -1,9 +1,36 @@
 FROM python:3.12
+
 WORKDIR /app
 EXPOSE 80
 
-COPY . .
+COPY requirements.txt requirements.txt
 
 RUN pip install --no-cache-dir -r requirements.txt
 
+# Below is an example command to run to set these environment variables needed for default secrets function to be set
+# For more variables, just extend the command with --build-arg <ARG name>=<your value>
+# For more info, see https://docs.docker.com/build/building/variables/#arg-usage-example
+# docker build --build-arg SECRET_PATH="/sample/app/test" --build-arg SECRET_ENCRYPTION_KEY_PATH="/sample/app/test/encrypt" --build-arg SECRET_CERT_PATH="/sample/app/test/cert" --build-arg SECRET_KEY_PATH="/sample/app/test/key" --build-arg ROLE_ARN="arn:aws:iam::767397936445:role/SampleAppRetrieveSecret" --build-arg REGION_NAME="ap-southeast-1" -t ssg/sample-app-test .
+
+ARG SECRET_PATH=''
+ENV SECRET_PATH=$SECRET_PATH
+
+ARG SECRET_ENCRYPTION_KEY_PATH=''
+ENV SECRET_ENCRYPTION_KEY_PATH=$SECRET_ENCRYPTION_KEY_PATH
+
+ARG SECRET_CERT_PATH=''
+ENV SECRET_CERT_PATH=$SECRET_CERT_PATH
+
+ARG SECRET_KEY_PATH=''
+ENV SECRET_KEY_PATH=$SECRET_KEY_PATH
+
+ARG ROLE_ARN=''
+ENV ROLE_ARN=$ROLE_ARN
+
+ARG REGION_NAME=''
+ENV REGION_NAME=$REGION_NAME
+
+
+COPY . .
+
 ENTRYPOINT ["streamlit", "run", "Home.py"]

SSG-API-Testing-Application-v2/app/Home.py

@@ -20,6 +20,8 @@
 from app.core.system.logger import Logger  # noqa: E402
 from app.core.constants import Endpoints  # noqa: E402
 
+from app.core.system.secrets import (ENV_NAME_ENCRYPT, ENV_NAME_CERT, ENV_NAME_KEY)  # noqa: E402
+
 # initialise all variables and logger
 init()
 LOGGER = Logger("Home")
@@ -31,11 +33,13 @@
 
 with st.sidebar:
     st.header("View Configs")
-    st.markdown("Click the `Configs` button to view your loaded configurations at any time!")
+    st.markdown(
+        "Click the `Configs` button to view your loaded configurations at any time!")
 
     if st.button("Configs", key="config_display", type="primary"):
         display_config()
 
+
 st.image("assets/sf.png", width=200)
 st.title("SSG API Sample Application")
 st.markdown("Welcome to the SSG API Sample Application!\n\n"
@@ -72,7 +76,15 @@
     else:
         LOGGER.info("UEN loaded!")
         st.success("**UEN** loaded successfully!", icon="✅")
-        st.session_state.update(uen=st.session_state["uen"].upper())  # UENs only have upper case characters
+        # UENs only have upper case characters
+        st.session_state.update(uen=st.session_state["uen"].upper())
+
+st.checkbox("Tick this if you would like to use our sample Encryption key, Certificate and Private Key instead",
+            key="default_secrets_checkbox",
+            help="This is a reminder that you need to have your own credentials when using the APIs in production")
+# logic here because streamlit will delete the session state when navigating to new page
+if st.session_state["default_secrets_checkbox"] is not None:
+    st.session_state["default_secrets"] = st.session_state["default_secrets_checkbox"]
 
 # AES Encryption Key to be loaded outside of a form
 st.session_state["encryption_key"] = st.text_input("Enter in your encryption key", type="password",
@@ -89,7 +101,8 @@
     else:
         LOGGER.info("Encryption Key loaded!")
         st.success("**Encryption Key** loaded successfully!", icon="✅")
-        st.session_state.update(encryption_key=st.session_state["encryption_key"])
+        st.session_state.update(
+            encryption_key=st.session_state["encryption_key"])
 
 # Credentials need to be loaded in a form to ensure that the pair submitted is valid
 with st.form(key="init_config"):
@@ -115,12 +128,14 @@
             try:
                 LOGGER.info("Verifying configurations...")
                 # save the byte stream into a temp file to give it a path for passing it to requests
-                st.session_state["cert_pem"] = NamedTemporaryFile(delete=False, delete_on_close=False, suffix=".pem")
+                st.session_state["cert_pem"] = NamedTemporaryFile(
+                    delete=False, delete_on_close=False, suffix=".pem")
                 st.session_state["cert_pem"].write(cert_pem.read())
                 st.session_state["cert_pem"] = st.session_state["cert_pem"].name
                 LOGGER.info("Certificate loaded!")
 
-                st.session_state["key_pem"] = NamedTemporaryFile(delete=False, delete_on_close=False, suffix=".pem")
+                st.session_state["key_pem"] = NamedTemporaryFile(
+                    delete=False, delete_on_close=False, suffix=".pem")
                 st.session_state["key_pem"].write(key_pem.read())
                 st.session_state["key_pem"] = st.session_state["key_pem"].name
                 LOGGER.info("Private key loaded!")
@@ -131,9 +146,11 @@
                     raise AssertionError("Certificate and private key are not valid! Are you sure that you "
                                          "have uploaded your certificates and private keys properly?")
                 LOGGER.info("Certificate and key verified!")
-                st.success("**Certificate and Key loaded successfully!**\n\n", icon="✅")
+                st.success(
+                    "**Certificate and Key loaded successfully!**\n\n", icon="✅")
             except base64.binascii.Error:
-                LOGGER.error("Certificate/Private key is not encoded in Base64, or that the cert/key is invalid!")
+                LOGGER.error(
+                    "Certificate/Private key is not encoded in Base64, or that the cert/key is invalid!")
                 st.error("Certificate or private key is invalid!", icon="🚨")
             except AssertionError as ex:
                 LOGGER.error(ex)

SSG-API-Testing-Application-v2/app/core/assessments/create_assessment.py

@@ -41,11 +41,11 @@ def _prepare(self, assessment_info: CreateAssessmentInfo) -> None:
             .with_header("Content-Type", "application/json") \
             .with_body(assessment_info.payload())
 
-    def execute(self) -> requests.Response:
+    def execute(self, encryption_key, cert_pem, key_pem) -> requests.Response:
         """
-        Executes the HTTP request and returns the response object
+        Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.post_encrypted()
+        return self.req.post_encrypted(encryption_key, cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/assessments/search_assessment.py

@@ -42,11 +42,11 @@ def _prepare(self, search_info: SearchAssessmentInfo):
             .with_header("Content-Type", "application/json") \
             .with_body(search_info.payload())
 
-    def execute(self) -> requests.Response:
+    def execute(self, encryption_key, cert_pem, key_pem) -> requests.Response:
         """
-        Executes the HTTP request and returns the response object
+        Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.post_encrypted()
+        return self.req.post_encrypted(encryption_key, cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/assessments/update_void_assessment.py

@@ -44,11 +44,11 @@ def _prepare(self, assessment_reference_number: str, assessment_info: UpdateVoid
             .with_header("Content-Type", "application/json") \
             .with_body(assessment_info.payload())
 
-    def execute(self) -> requests.Response:
+    def execute(self, encryption_key, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.post_encrypted()
+        return self.req.post_encrypted(encryption_key, cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/assessments/view_assessment.py

@@ -39,11 +39,11 @@ def _prepare(self, referenceNumber: str) -> None:
             .with_header("accept", "application/json") \
             .with_header("Content-Type", "application/json")
 
-    def execute(self) -> requests.Response:
+    def execute(self, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.get()
+        return self.req.get(cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/attendance/course_session_attendance.py

@@ -46,11 +46,11 @@ def _prepare(self, runId: int, crn: str, session_id: str) -> None:
             .with_param("courseReferenceNumber", crn) \
             .with_param("sessionId", session_id)
 
-    def execute(self) -> requests.Response:
+    def execute(self, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.get()
+        return self.req.get(cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/attendance/upload_course_session_attendance.py

@@ -45,11 +45,11 @@ def _prepare(self, runId: int, attendanceInfo: UploadAttendanceInfo) -> None:
             .with_header("Content-Type", "application/json") \
             .with_body(attendanceInfo.payload())
 
-    def execute(self) -> requests.Response:
+    def execute(self, encryption_key, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.post_encrypted()
+        return self.req.post_encrypted(encryption_key, cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/cipher/encrypt_decrypt.py

@@ -19,32 +19,35 @@ class Cryptography:
     INITIAL_VECTOR: bytes = "SSGAPIInitVector".encode()
 
     @staticmethod
-    def encrypt(plaintext: bytes | str, return_bytes: bool = True, key: str = None) -> bytes | str | None:
+    def encrypt(key: str, plaintext: bytes | str, return_bytes: bool = True) -> bytes | str | None:
         """
         Encrypts a message using AES-256/CBC/PKCS7 and returns the ciphertext.
 
+        :param key: Encryption key to encrypt the plaintext
         :param plaintext: Plaintext Message to be encrypted. If a string is passed as the argument, it will be
                           encoded into a bytes object.
         :param return_bytes: If True, the ciphertext will be returned as a bytes object.
-        :param key: Key to override key stored in session state
         :return: Ciphertext
         """
 
-        if ("encryption_key" not in st.session_state
-                or st.session_state["encryption_key"] is None
-                or len(st.session_state["encryption_key"]) == 0) and not key:
-            # if there are no keys loaded, do not continue
-            raise AttributeError("No encryption key loaded!")
+        # check if encryption key is set in session state
+        # if ("encryption_key" not in st.session_state
+        #         or st.session_state["encryption_key"] is None
+        #         or len(st.session_state["encryption_key"]) == 0) and not key:
+        #     # if there are no keys loaded, do not continue
+        #     raise AttributeError("No encryption key loaded!")
 
         if isinstance(plaintext, str):
             plaintext = plaintext.encode()
 
-        enc_key = b64decode(key if key else st.session_state["encryption_key"])
-        cipher_algo = Cipher(AES(enc_key), CBC(Cryptography.INITIAL_VECTOR), backend=default_backend())
+        enc_key = b64decode(key)
+        cipher_algo = Cipher(AES(enc_key), CBC(
+            Cryptography.INITIAL_VECTOR), backend=default_backend())
         padding_algo = PKCS7(128).padder()
 
         encryptor = cipher_algo.encryptor()
-        padded_plaintext = padding_algo.update(plaintext) + padding_algo.finalize()
+        padded_plaintext = padding_algo.update(
+            plaintext) + padding_algo.finalize()
         ciphertext = encryptor.update(padded_plaintext) + encryptor.finalize()
 
         encoded_ciphertext = b64encode(ciphertext)
@@ -55,33 +58,36 @@ def encrypt(plaintext: bytes | str, return_bytes: bool = True, key: str = None)
         return encoded_ciphertext.decode()
 
     @staticmethod
-    def decrypt(ciphertext: str | bytes, return_bytes: bool = True, key: str = None) -> bytes | str | None:
+    def decrypt(key: str, ciphertext: str | bytes, return_bytes: bool = True) -> bytes | str | None:
         """
         Decrypts an encrypted message and returns the plaintext.
 
+        :param key: Key to decrypt ciphertext
         :param ciphertext: Ciphertext Message to be decrypted. It does not matter if a string or bytes are provided,
                            both will be encoded into a bytes object with base64-decode.
-        :param key: Key to override key stored in session state
         :param return_bytes: If True, the ciphertext will be returned as a bytes object.
         :return: Plaintext Message
         """
 
-        if ("encryption_key" not in st.session_state
-                or st.session_state["encryption_key"] is None
-                or len(st.session_state["encryption_key"]) == 0) and not key:
-            # if there are no keys loaded, do not continue
-            return None
+        # check if encryption key is set in session state
+        # if ("encryption_key" not in st.session_state
+        #         or st.session_state["encryption_key"] is None
+        #         or len(st.session_state["encryption_key"]) == 0) and not key:
+        #     # if there are no keys loaded, do not continue
+        #     return None
 
         # decode the input text into a bytes object
         decoded_ciphertext = b64decode(ciphertext)
 
-        enc_key = b64decode(key if key else st.session_state["encryption_key"])
-        cipher_algo = Cipher(AES(enc_key), CBC(Cryptography.INITIAL_VECTOR), backend=default_backend())
+        enc_key = b64decode(key)
+        cipher_algo = Cipher(AES(enc_key), CBC(
+            Cryptography.INITIAL_VECTOR), backend=default_backend())
         padding_algo = PKCS7(128).unpadder()
 
         decryptor = cipher_algo.decryptor()
         plaintext = decryptor.update(decoded_ciphertext) + decryptor.finalize()
-        unpadded_plaintext = padding_algo.update(plaintext) + padding_algo.finalize()
+        unpadded_plaintext = padding_algo.update(
+            plaintext) + padding_algo.finalize()
 
         if return_bytes:
             return unpadded_plaintext

SSG-API-Testing-Application-v2/app/core/courses/add_course_run.py

@@ -46,15 +46,16 @@ def _prepare(self, include_expired: OptionalSelector, runinfo: AddRunInfo) -> No
             case OptionalSelector.YES:
                 self.req = self.req.with_param("includeExpiredCourses", "true")
             case OptionalSelector.NO:
-                self.req = self.req.with_param("includeExpiredCourses", "false")
+                self.req = self.req.with_param(
+                    "includeExpiredCourses", "false")
 
         self.req = self.req.with_body(runinfo.payload())
 
-    def execute(self) -> requests.Response:
+    def execute(self, encryption_key, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.post_encrypted()
+        return self.req.post_encrypted(encryption_key, cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/courses/delete_course_run.py

@@ -53,11 +53,11 @@ def _prepare(self, runId: str, include_expired: OptionalSelector, delete_runinfo
 
         self.req = self.req.with_body(delete_runinfo.payload())
 
-    def execute(self) -> requests.Response:
+    def execute(self, encryption_key, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.post_encrypted()
+        return self.req.post_encrypted(encryption_key, cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/courses/edit_course_run.py

@@ -49,15 +49,16 @@ def _prepare(self, runId: str, include_expired: OptionalSelector, runinfo: EditR
             case OptionalSelector.YES:
                 self.req = self.req.with_param("includeExpiredCourses", "true")
             case OptionalSelector.NO:
-                self.req = self.req.with_param("includeExpiredCourses", "false")
+                self.req = self.req.with_param(
+                    "includeExpiredCourses", "false")
 
         self.req = self.req.with_body(runinfo.payload())
 
-    def execute(self) -> requests.Response:
+    def execute(self, encryption_key, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.post_encrypted()
+        return self.req.post_encrypted(encryption_key, cert_pem, key_pem)

SSG-API-Testing-Application-v2/app/core/courses/view_course_run.py

@@ -47,13 +47,14 @@ def _prepare(self, runId: str, include_expired: OptionalSelector) -> None:
             case OptionalSelector.YES:
                 self.req = self.req.with_param("includeExpiredCourses", "true")
             case OptionalSelector.NO:
-                self.req = self.req.with_param("includeExpiredCourses", "false")
+                self.req = self.req.with_param(
+                    "includeExpiredCourses", "false")
 
-    def execute(self) -> requests.Response:
+    def execute(self, cert_pem, key_pem) -> requests.Response:
         """
         Executes the HTTP request and returns the response object.
 
         :return: requests.Response object
         """
 
-        return self.req.get()
+        return self.req.get(cert_pem, key_pem)