feat(backend): Allow project auto-claim with pre-defined token

doc/getting-started.md

@@ -77,6 +77,11 @@ Using `docatl`:
 docatl claim awesome-project --host http://localhost:8000
 ```
 
+In addition, it is possible to claim all projects with a token previously defined via environment variables
+instead of generating a new token for each project. This is particularly useful for CI/CD applications where the
+documentation is automatically uploaded by a build server. This feature needs to be enabled in the backend. You can
+read more about this feature in the [backend documentation](../docat/README.md).
+
 #### Authentication
 
 To make an authenticated call, specify a header with the key `Docat-Api-Key` and your token as the value:
@@ -170,4 +175,4 @@ Using `docatl`:
 
 ```sh
 docatl show awesome-project 0.0.1 --host http://localhost:8000 --api-key <token>
-```
+```

docat/README.md

@@ -21,6 +21,8 @@ poetry install
 * **DOCAT_SERVE_FILES**: Serve static documentation instead of a nginx (for testing)
 * **DOCAT_STORAGE_PATH**: Upload directory for static files (needs to match nginx config)
 * **FLASK_DEBUG**: Start flask in debug mode
+* **DOCAT_GLOBAL_CLAIM_TOKEN**: If set, all uploaded projects are being claimed automatically with that token.
+* **DOCAT_GLOBAL_CLAIM_SALT**: If set, all claims will be made with this salt.
 
 ## Usage
 

docat/docat/app.py

@@ -8,22 +8,23 @@
 :license: MIT, see LICENSE for more details.
 """
 import os
-import secrets
 import shutil
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Union
 
 import magic
 from fastapi import Depends, FastAPI, File, Header, Response, UploadFile, status
 from fastapi.staticfiles import StaticFiles
 from starlette.responses import JSONResponse
 from tinydb import Query, TinyDB
 
+from docat.constants import get_global_claim_token
 from docat.models import ApiResponse, ClaimResponse, ProjectDetail, Projects, TokenStatus
 from docat.utils import (
     DB_PATH,
     UPLOAD_FOLDER,
     calculate_token,
+    claim_project,
     create_symlink,
     extract_archive,
     get_all_projects,
@@ -197,15 +198,16 @@ def show_version(
     return ApiResponse(message=f"Version {version} is now shown")
 
 
-@app.post("/api/{project}/{version}", response_model=ApiResponse, status_code=status.HTTP_201_CREATED)
-@app.post("/api/{project}/{version}/", response_model=ApiResponse, status_code=status.HTTP_201_CREATED)
+@app.post("/api/{project}/{version}", response_model=Union[ApiResponse, ClaimResponse], status_code=status.HTTP_201_CREATED)
+@app.post("/api/{project}/{version}/", response_model=Union[ApiResponse, ClaimResponse], status_code=status.HTTP_201_CREATED)
 def upload(
     project: str,
     version: str,
     response: Response,
     file: UploadFile = File(...),
     docat_api_key: Optional[str] = Header(None),
     db: TinyDB = Depends(get_db),
+    global_claim_token: bool = Depends(get_global_claim_token),
 ):
     if is_forbidden_project_name(project):
         response.status_code = status.HTTP_400_BAD_REQUEST
@@ -241,11 +243,24 @@ def upload(
         shutil.copyfileobj(file.file, buffer)
 
     extract_archive(target_file, base_path)
+    index_file_exists = (base_path / "index.html").exists()
 
-    if not (base_path / "index.html").exists():
-        return ApiResponse(message="Documentation uploaded successfully, but no index.html found at root of archive.")
+    token: Optional[str] = global_claim_token or docat_api_key
+    if token is not None:
+        token = claim_project(project=project, db=db, token=token)
+        if index_file_exists:
+            message = "Documentation uploaded and claimed successfully"
+        else:
+            message = "Documentation uploaded and claimed successfully, but no index.html found at root of archive."
+    else:
+        if index_file_exists:
+            message = "Documentation uploaded successfully"
+        else:
+            message = "Documentation uploaded successfully, but no index.html found at root of archive."
 
-    return ApiResponse(message="Documentation uploaded successfully")
+    if token:
+        return ClaimResponse(message=message, token=token)
+    return ApiResponse(message=message)
 
 
 @app.put("/api/{project}/{version}/tags/{new_tag}", response_model=ApiResponse, status_code=status.HTTP_201_CREATED)
@@ -278,18 +293,14 @@ def tag(project: str, version: str, new_tag: str, response: Response):
     responses={status.HTTP_409_CONFLICT: {"model": ApiResponse}},
 )
 def claim(project: str, db: TinyDB = Depends(get_db)):
-    Project = Query()
-    table = db.table("claims")
-    result = table.search(Project.name == project)
-    if result:
-        return JSONResponse(status_code=status.HTTP_409_CONFLICT, content={"message": f"Project {project} is already claimed!"})
-
-    token = secrets.token_hex(16)
-    salt = os.urandom(32)
-    token_hash = calculate_token(token, salt)
-    table.insert({"name": project, "token": token_hash, "salt": salt.hex()})
-
-    return ClaimResponse(message=f"Project {project} successfully claimed", token=token)
+    try:
+        token = claim_project(project=project, db=db)
+        return ClaimResponse(message=f"Project {project} successfully claimed", token=token)
+    except PermissionError as error:
+        return JSONResponse(
+            status_code=status.HTTP_409_CONFLICT,
+            content={"message": str(error)},
+        )
 
 
 @app.put("/api/{project}/rename/{new_project_name}", response_model=ApiResponse, status_code=status.HTTP_200_OK)

docat/docat/constants.py

@@ -0,0 +1,26 @@
+import os
+from typing import Optional
+
+ENV_GLOBAL_CLAIM_TOKEN = "DOCAT_GLOBAL_CLAIM_TOKEN"
+ENV_GLOBAL_CLAIM_SALT = "DOCAT_GLOBAL_CLAIM_SALT"
+
+
+def get_global_claim_token() -> Optional[str]:
+    """Returns the global claim token which can be defined by an environment variable.
+
+    Returns:
+        The optional global claim token or None.
+    """
+    return os.environ.get(ENV_GLOBAL_CLAIM_TOKEN, None)
+
+
+def get_global_claim_salt() -> Optional[bytes]:
+    """Returns the global claim salt which can be defined by an environment variable.
+
+    Returns:
+        The optional global claim salt or None.
+    """
+    global_claim_salt = os.environ.get(ENV_GLOBAL_CLAIM_SALT, None)
+    if global_claim_salt is not None:
+        return global_claim_salt.encode()
+    return None

docat/docat/utils.py

@@ -3,10 +3,15 @@
 """
 import hashlib
 import os
+import secrets
 import shutil
 from pathlib import Path
+from typing import Optional
 from zipfile import ZipFile
 
+from tinydb import Query, TinyDB
+
+from docat.constants import get_global_claim_salt, get_global_claim_token
 from docat.models import Project, ProjectDetail, Projects, ProjectVersion
 
 NGINX_CONFIG_PATH = Path("/etc/nginx/locations.d")
@@ -162,3 +167,38 @@ def should_include(name: str) -> bool:
             reverse=True,
         ),
     )
+
+
+def claim_project(project: str, db: TinyDB, token: Optional[str] = None) -> str:
+    """Claims a project.
+
+    The token used for claiming is determined as follows.
+
+        1. If a token has been sent in the upload request, this will be used.
+        2. Otherwise, the system checks whether the global token is set.
+        3. If both cases do not apply, a randomly generated token is used.
+
+    Args:
+        project: The project name.
+        db: The database to use.
+        token: The optional token to use for claiming the project.
+
+    Raises:
+        PermissionError: If the project has already been claimed.
+
+    Returns:
+        The claim token.
+    """
+    table = db.table("claims")
+
+    # Check if the project has already been claimed
+    if table.search(Query().name == project):
+        raise PermissionError(f"Project {project} is already claimed!")
+
+    _token = token or get_global_claim_token() or secrets.token_hex(16)
+    _salt = get_global_claim_salt() or os.urandom(32)
+
+    token_hash = calculate_token(_token, _salt)
+    table.insert({"name": project, "token": token_hash, "salt": _salt.hex()})
+
+    return _token

docat/tests/test_upload.py

@@ -1,8 +1,11 @@
 import io
+import os
 from pathlib import Path
 from unittest.mock import call, patch
 
 import docat.app as docat
+from docat.app import check_token_for_project
+from docat.constants import ENV_GLOBAL_CLAIM_SALT, ENV_GLOBAL_CLAIM_TOKEN
 
 
 def test_successfully_upload(client):
@@ -15,6 +18,39 @@ def test_successfully_upload(client):
         assert (docat.DOCAT_UPLOAD_FOLDER / "some-project" / "1.0.0" / "index.html").exists()
 
 
+@patch.dict(os.environ, {ENV_GLOBAL_CLAIM_SALT: "test-salt", ENV_GLOBAL_CLAIM_TOKEN: "test-token"})
+def test_successfully_upload_with_global_claim(client):
+    with patch("docat.app.remove_docs"):
+        response = client.post("/api/some-project/1.0.0", files={"file": ("index.html", io.BytesIO(b"<h1>Hello World</h1>"), "plain/text")})
+        response_data = response.json()
+
+        assert response.status_code == 201
+        assert response_data["message"] == "Documentation uploaded and claimed successfully"
+        assert "test-token" == response_data["token"]
+        assert (docat.DOCAT_UPLOAD_FOLDER / "some-project" / "1.0.0" / "index.html").exists()
+
+    status = check_token_for_project(db=docat.db, token="test-token", project="some-project")
+    assert True is status.valid
+
+
+def test_successfully_upload_with_header_token(client):
+    with patch("docat.app.remove_docs"):
+        response = client.post(
+            "/api/some-project/1.0.0",
+            files={"file": ("index.html", io.BytesIO(b"<h1>Hello World</h1>"), "plain/text")},
+            headers={"Docat-Api-Key": "1234"},
+        )
+        response_data = response.json()
+
+        assert response.status_code == 201
+        assert response_data["message"] == "Documentation uploaded and claimed successfully"
+        assert "1234" == response_data["token"]
+        assert (docat.DOCAT_UPLOAD_FOLDER / "some-project" / "1.0.0" / "index.html").exists()
+
+    status = check_token_for_project(db=docat.db, token="1234", project="some-project")
+    assert True is status.valid
+
+
 def test_successfully_override(client_with_claimed_project):
     with patch("docat.app.remove_docs") as remove_mock:
         response = client_with_claimed_project.post(