Add API endpoint to download exports (CTFd#2547)

* Add `POST /api/v1/exports/raw` to download CTFd and CSV exports via the API

Author: ColdHeat

CTFd/api/__init__.py

@@ -6,6 +6,7 @@
 from CTFd.api.v1.challenges import challenges_namespace
 from CTFd.api.v1.comments import comments_namespace
 from CTFd.api.v1.config import configs_namespace
+from CTFd.api.v1.exports import exports_namespace
 from CTFd.api.v1.files import files_namespace
 from CTFd.api.v1.flags import flags_namespace
 from CTFd.api.v1.hints import hints_namespace
@@ -75,3 +76,4 @@
 CTFd_API_v1.add_namespace(comments_namespace, "/comments")
 CTFd_API_v1.add_namespace(shares_namespace, "/shares")
 CTFd_API_v1.add_namespace(brackets_namespace, "/brackets")
+CTFd_API_v1.add_namespace(exports_namespace, "/exports")

CTFd/api/v1/exports.py

@@ -0,0 +1,46 @@
+from datetime import datetime as DateTime
+
+from flask import request, send_file
+from flask_restx import Namespace, Resource
+
+from CTFd.utils.config import ctf_name
+from CTFd.utils.csv import dump_csv
+from CTFd.utils.decorators import admins_only, ratelimit
+from CTFd.utils.exports import export_ctf as export_ctf_util
+
+exports_namespace = Namespace("exports", description="Endpoint to retrieve Exports")
+
+
+@exports_namespace.route("/raw")
+class ExportList(Resource):
+    @admins_only
+    @ratelimit(method="POST", limit=10, interval=60)
+    def post(self):
+        req = request.get_json()
+        export_type = req.get("type", "_")
+        export_args = req.get("args", {})
+
+        day = DateTime.now().strftime("%Y-%m-%d_%T")
+        if export_type == "csv":
+            table = export_args.get("table")
+            if not table:
+                return {
+                    "success": False,
+                    "errors": {"args": "Missing table to export"},
+                }, 400
+            output = dump_csv(name=table)
+            return send_file(
+                output,
+                as_attachment=True,
+                max_age=-1,
+                download_name=f"{ctf_name()}-{table}-{day}.csv",
+            )
+        else:
+            backup = export_ctf_util()
+            full_name = f"{ctf_name()}.{day}.zip"
+            return send_file(
+                backup,
+                cache_timeout=-1,
+                as_attachment=True,
+                attachment_filename=full_name,
+            )

CTFd/utils/exports/__init__.py

@@ -6,7 +6,7 @@
 import sys
 import tempfile
 import zipfile
-from io import BytesIO
+from io import BytesIO, StringIO
 from pathlib import Path
 
 import dataset
@@ -62,7 +62,7 @@ def export_ctf():
             "results": [{"version_num": get_current_revision()}],
             "meta": {},
         }
-        result_file = BytesIO()
+        result_file = StringIO()
         json.dump(result, result_file)
         result_file.seek(0)
         backup_zip.writestr("db/alembic_version.json", result_file.read())

tests/api/v1/test_exports.py

@@ -0,0 +1,49 @@
+from tests.helpers import (
+    create_ctfd,
+    destroy_ctfd,
+    gen_challenge,
+    login_as_user,
+    register_user,
+)
+
+
+def test_api_export_csv():
+    app = create_ctfd()
+    with app.app_context():
+        gen_challenge(app.db)
+        data = {
+            "type": "csv",
+            "args": {"table": "challenges"},
+        }
+
+        with login_as_user(app, name="admin", password="password") as client:
+            r = client.post("/api/v1/exports/raw", json=data)
+            assert r.status_code == 200
+            assert r.headers["Content-Type"].startswith("text/csv")
+            assert "chal_name" in r.get_data(as_text=True)
+
+        # Test that regular users cannot access the endpoint
+        register_user(app)
+        with login_as_user(app) as client:
+            response = client.post("/api/v1/exports/raw", json=data)
+            assert response.status_code == 403
+    destroy_ctfd(app)
+
+
+def test_api_export():
+    app = create_ctfd()
+    with app.app_context():
+        gen_challenge(app.db)
+        data = {}
+
+        with login_as_user(app, name="admin", password="password") as client:
+            r = client.post("/api/v1/exports/raw", json=data)
+            assert r.status_code == 200
+            assert r.headers["Content-Type"].startswith("application/zip")
+
+        # Test that regular users cannot access the endpoint
+        register_user(app)
+        with login_as_user(app) as client:
+            response = client.post("/api/v1/exports/raw", json=data)
+            assert response.status_code == 403
+    destroy_ctfd(app)