ci: add functional tests

ctfer-io · Sep 4, 2024 · 94fec49 · 94fec49
1 parent 1c32389
commit 94fec49
Show file tree

Hide file tree

Showing 3 changed files with 264 additions and 13 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -14,27 +14,18 @@ jobs:
         image: ctferio/chall-manager:dev
         ports:
           - 9090:9090
-          - 8081:8081
         env:
-          CI: true
+          GATEWAY: true
         credentials:
           username: ${{ secrets.docker_username }}
           password: ${{ secrets.docker_password }}
-
-      mariadb:
-        image: mariadb:10.7.8
-        ports:
-          - 3306:3306
-        env:
-          MYSQL_ROOT_PASSWORD: ctfer
 
       ctfd:
         image: ctfd/ctfd:3.7.3@sha256:90470e1fe0f93028ce6ac197b8942916ee157d4b5d33c8266c5bec7662e55ac3
         ports:
           - 8000:8000
         options: --mount type=bind,source=${{ github.workspace }},target=/opt/CTFd/CTFd/plugins/ctfd-chall-manager
         env:
-          DATABASE_URL: mysql+pymysql://root:ctfer@mariadb/ctfd
           LOG_LEVEL: DEBUG
 
     steps:
@@ -91,6 +82,9 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@0a12ed9d6a96ab950c8f026ed9f722fe0da7ef32 # v5.0.2
 
+      - name: Setup Python
+        uses: actions/setup-python@f677139bbe7f9c59b41e40162b753c062f5d49a3 # v5.2.0
+
       - name: Generate token and store it for GH Actions
         run: |-
           cd hack/token  
@@ -138,4 +132,47 @@ jobs:
         with:
           name: cypress-screenshots
           path: cypress/screenshots
-          if-no-files-found: ignore
+          if-no-files-found: ignore
+
+      - name: Generate token and store it for GH Actions
+        run: |-
+          cd hack/token  
+          go run main.go
+        env:
+          CTFD_URL: http://localhost:8000
+          CTFD_NAME: "user1"
+          CTFD_PASSWORD: "user1"
+          CTFD_OUTPUT_KEY: CTFD_API_TOKEN_USER
+
+      - name: Generate token and store it for GH Actions
+        run: |-
+          cd hack/token  
+          go run main.go
+        env:
+          CTFD_URL: http://localhost:8000
+          CTFD_NAME: ${{ env.NAME }}
+          CTFD_PASSWORD: ${{ env.PASSWORD }}
+          CTFD_OUTPUT_KEY: CTFD_API_TOKEN_ADMIN
+
+      - name: Run Fonctional Tests in mode Teams
+        run: |- 
+          cd test
+          python -m unittest test_api.py
+        env:
+          CTFD_URL: http://localhost:8000
+
+      - name: Update CTFd into mode users
+        uses: ctfer-io/ctfd-setup@f76c764b07af9c6292de3f75009876e26c77bb1f # v1.2.1
+        with:
+          url: 'http://ctfd:8000'
+          mode: users 
+          admin_name: ${{ env.NAME }}
+          admin_email: [email protected]
+          admin_password: ${{ env.PASSWORD }}
+
+      - name: Run Fonctional Tests in mode Users
+        run: |- 
+          cd test
+          python -m unittest test_api.py
+        env:
+          CTFD_URL: http://localhost:8000
diff --git a/hack/token/main.go b/hack/token/main.go
@@ -12,6 +12,10 @@ import (
 // and creates an API key ready to work.
 
 func main() {
+	key := "CTFD_API_KEY"
+	if k, ok := os.LookupEnv("CTFD_OUTPUT_KEY"); ok {
+		key = k
+	}
 	url := os.Getenv("CTFD_URL")
 	name := os.Getenv("CTFD_NAME")
 	password := os.Getenv("CTFD_PASSWORD")
@@ -47,7 +51,7 @@ func main() {
 		log.Fatalf("Opening $GITHUB_ENV file (%s): %s", ghf, err)
 	}
 	defer f.Close()
-	if _, err := f.WriteString(fmt.Sprintf("CTFD_API_KEY=%s\n", *token.Value)); err != nil {
-		log.Fatalf("Writing CTFD_API_KEY to $GITHUB_ENV file (%s): %s", ghf, err)
+	if _, err := f.WriteString(fmt.Sprintf("%s=%s\n", key, *token.Value)); err != nil {
+		log.Fatalf("Writing %s to $GITHUB_ENV file (%s): %s", key, ghf, err)
 	}
 }
diff --git a/test/test_api.py b/test/test_api.py
@@ -0,0 +1,210 @@
+import unittest
+import requests
+import json
+import os 
+
+ctfd_url = os.getenv("CTFD_URL")
+base_url = f"{ctfd_url}/api/v1/plugins/ctfd-chall-manager"
+ctfd_token_user = os.getenv("CTFD_API_TOKEN_USER")
+ctfd_token_admin = os.getenv("CTFD_API_TOKEN_ADMIN")
+
+headers_user = {
+    "Accept": "application/json",
+    "Authorization": f"Token {ctfd_token_user}",
+    "Content-Type": "application/json"
+}
+
+headers_admin = {
+    "Accept": "application/json",
+    "Authorization": f"Token {ctfd_token_admin}",
+    "Content-Type": "application/json"
+}
+
+def post_instance(challengeId: int):
+    payload = {
+        "challengeId": f"{challengeId}"
+    }
+    r = requests.post(f"{base_url}/instance", headers=headers_user, data=json.dumps(payload))
+    return r
+
+def get_instance(challengeId: int):
+    r = requests.get(f"{base_url}/instance?challengeId={challengeId}",  headers=headers_user)
+    return r
+
+def delete_instance(challengeId: int):
+    r = requests.delete(f"{base_url}/instance?challengeId={challengeId}",  headers=headers_user)
+    return r
+
+def patch_instance(challengeId: int): 
+    r = requests.patch(f"{base_url}/instance?challengeId={challengeId}",  headers=headers_user)
+    return r
+
+class Test_F_UserMana(unittest.TestCase):
+    def test_valid_get(self):
+        r = requests.get(f"{base_url}/mana",  headers=headers_user)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+class Test_F_UserInstance(unittest.TestCase):
+    def test_create_patch_delete_visible_challenge(self):
+        r = get_instance(1)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+        r = post_instance(1)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)        
+        self.assertEqual("connectionInfo" in a["data"]["message"].keys(), True)
+
+        r = get_instance(1)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+        self.assertEqual("connectionInfo" in a["data"]["message"].keys(), True)
+
+        r = patch_instance(1)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+        r = delete_instance(1)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+    def test_create_invalid_challenge(self):
+        r = post_instance(9999)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+    def test_delete_invalid_challenge(self):
+        r = delete_instance(9999)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+    def test_delete_valid_challenge_but_no_instance(self):
+        r = delete_instance(1)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+    def test_patch_valid_challenge_but_no_instance(self):
+        r = patch_instance(1)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+    def test_get_missing_arg(self):
+        r = requests.get(f"{base_url}/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 400)
+
+    def test_post_missing_arg(self):
+        r = requests.post(f"{base_url}/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 400)
+
+    def test_patch_missing_arg(self):
+        r = requests.patch(f"{base_url}/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 400)
+
+    def test_delete_missing_arg(self):
+        r = requests.delete(f"{base_url}/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 400)
+
+class Test_F_AdminInstance(unittest.TestCase):
+    def test_user_connection_is_denied(self):
+        r = requests.get(f"{base_url}/admin/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 403)
+        r = requests.post(f"{base_url}/admin/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 403)
+        r = requests.patch(f"{base_url}/admin/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 403)
+        r = requests.delete(f"{base_url}/admin/instance",  headers=headers_user)
+        self.assertEqual(r.status_code, 403)
+
+    def test_valid_challenge_valid_source(self):
+        challengeId = 1
+        sourceId = 1
+
+        r = requests.get(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+        payload = {
+            "challengeId": f"{challengeId}",
+            "sourceId": f"{sourceId}"
+        }
+        r = requests.post(f"{base_url}/admin/instance",  headers=headers_admin, data=json.dumps(payload))
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+        r = requests.get(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+        self.assertEqual("connectionInfo" in a["data"]["message"].keys(), True)
+
+        r = requests.patch(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+        r = requests.delete(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+    def test_invalid_challenge_valid_source(self):
+        challengeId = 999999
+        sourceId = 1
+
+        r = requests.get(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True) 
+
+        payload = {
+            "challengeId": f"{challengeId}",
+            "sourceId": f"{sourceId}"
+        }
+        r = requests.post(f"{base_url}/admin/instance",  headers=headers_admin, data=json.dumps(payload))
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+        r = requests.patch(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+        r = requests.delete(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+    def test_valid_challenge_unknown_source(self):
+        challengeId = 1
+        sourceId = 999999
+
+        r = requests.get(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True) 
+
+        payload = {
+            "challengeId": f"{challengeId}",
+            "sourceId": f"{sourceId}"
+        }
+        r = requests.post(f"{base_url}/admin/instance",  headers=headers_admin, data=json.dumps(payload))
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+        r = requests.patch(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+        r = requests.delete(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], True)
+
+    def test_delete_valid_challenge_but_no_instance(self):
+        challengeId = 1
+        sourceId = 999999
+        r = requests.delete(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+    def test_patch_valid_challenge_but_no_instance(self):
+        challengeId = 1
+        sourceId = 999999
+        r = requests.patch(f"{base_url}/admin/instance?challengeId={challengeId}&sourceId={sourceId}",  headers=headers_admin)
+        a = json.loads(r.text)
+        self.assertEqual(a["success"], False)
+
+# TODO add test_hidden_challenge()