Finish remaining code

.gitignore

@@ -0,0 +1,8 @@
+__pycache__/
+*.pyc
+.venv/
+.env
+.pytest_cache/
+*.egg-info/
+dist/
+build/

Dockerfile

@@ -0,0 +1,16 @@
+FROM docker.io/library/python:3.14-slim
+
+RUN groupadd -r appuser && useradd -r -g appuser appuser
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app/ ./app/
+
+USER appuser
+
+EXPOSE 5000
+
+CMD ["gunicorn", "-b", "0.0.0.0:5000", "app.server:app"]

README.md

@@ -1,12 +1,82 @@
 # jss-devsecops-challenge
 
-This is a practical test for a DevSecOps Engineer candidate at Bank J. Safra Sarasin. Instructions received (on 09.02.2025) from the recruiter are attached to this repo.
+DevSecOps assessment - device registration and statistics API built with Python (Flask) and MariaDB.
 
-### On 09.02.2026
+## What it does
-I created repo on https://git.harshanu.space. I'll work on code changes tomorrow as I will be free in the afternoon (Fashing Tuesday, half day work)
 
-### On 10.02.2026
+A single API server that tracks device type registrations per user. It exposes three endpoints:
-- Defined the database schema for the API endpoints in `init.sql`.
+
-- Pulled the `docker.io/library/mariadb:12` Docker image.
+- `POST /log/auth` - logs a user login event by storing the device type
-- Tested the schema by spinning up a MariaDB container and verifying the table creation.
+- `GET /log/auth/statistics?deviceType=iOS` - returns count of registrations for a given device type
-- Python will be used to write API endpoints
+- `POST /device/register` - directly registers a device type for a user
+
+Accepted device types: iOS, Android, Watch, TV, Tablet, Desktop, IoT.
+In contrast to API endpoints asked in pdf file, I used smaller case here.
+
+## Project structure
+
+```
+app/server.py           - Flask application with all endpoints
+tests/test_app.py       - unit tests (mocked DB)
+init.sql                - MariaDB schema initialization
+Dockerfile              - container image for the API
+docker-compose.yml      - full stack (MariaDB + API)
+requirements.txt        - Python dependencies
+```
+
+## Docker Hub
+
+The API image is published to Docker Hub:
+
+```
+docker.io/harshavmb/jss-devsecops-api:latest
+```
+
+To pull it directly:
+
+```bash
+docker pull harshavmb/jss-devsecops-api:latest
+```
+
+MariaDB is used from the official registry (`docker.io/library/mariadb:12`) and does not need a custom image. The schema is applied at startup via the init SQL script mounted in docker-compose.
+
+## Running locally with Docker Compose
+
+```bash
+docker compose up --build -d
+```
+
+This starts MariaDB, runs the init SQL script, and launches the API on port 5000.
+
+To stop:
+
+```bash
+docker compose down -v
+```
+
+## Running tests
+
+```bash
+pip install -r requirements.txt
+pytest tests/ -v
+```
+
+## Security controls
+
+- Input validation on userKey (regex whitelist) and deviceType (enum whitelist)
+- Parameterized SQL queries to prevent injection
+- Rate limiting (30 req/min per endpoint, 60 req/min global)
+- Non-root container user
+- No secrets hardcoded (configurable via DATABASE_URL env var)
+- No docker/podman container is run as root user (that depends on how docker/podman are configured on host)
+- Authentication of endpoints is left out for brevity
+
+## Configuration
+
+The API reads `DATABASE_URL` from the environment. Format:
+
+```
+mysql://user:password@host:port/database
+```
+
+Default: `mysql://root:my-secret-pw@localhost:3306/devsecops_db`

TIMELINE.md

@@ -0,0 +1,19 @@
+# jss-devsecops-challenge
+
+This is a practical test for a DevSecOps Engineer candidate at Bank J. Safra Sarasin. Instructions received (on 09.02.2025) from the recruiter are attached to this repo.
+
+### On 09.02.2026
+I created repo on https://git.harshanu.space. I'll work on code changes tomorrow as I will be free in the afternoon (Fashing Tuesday, half day work)
+
+### On 10.02.2026
+- Defined the database schema for the API endpoints in `init.sql`.
+- Pulled the `docker.io/library/mariadb:12` Docker image.
+- Tested the schema by spinning up a MariaDB container and verifying the table creation.
+- Python will be used to write API endpoints
+
+## On 13.02.2026
+- Finished the remaining part
+- app folder contains the python code
+- tests folder contains the unit tests
+- Dockerfile builds the python code in app/ directory & uses `gunicorn` HTTP web server to serve API endpoints
+- docker-compose.yml 

app/server.py

@@ -0,0 +1,121 @@
+import os
+import re
+from urllib.parse import urlparse
+import mysql.connector
+from flask import Flask, request, jsonify
+from flask_limiter import Limiter
+from flask_limiter.util import get_remote_address
+
+app = Flask(__name__)
+
+limiter = Limiter(get_remote_address, app=app, default_limits=["60 per minute"])
+
+DATABASE_URL = os.environ.get(
+    "DATABASE_URL", "mysql://root:my-secret-pw@localhost:3306/devsecops_db"
+)
+
+## I added more devices as I felt they are relevant to the ones asked in pdf
+VALID_DEVICE_TYPES = {"iOS", "Android", "Watch", "TV", "Tablet", "Desktop", "IoT"}
+USER_KEY_PATTERN = re.compile(r"^[a-zA-Z0-9_\-\.@]{1,255}$")
+
+
+def get_db_connection():
+    parsed = urlparse(DATABASE_URL)
+    return mysql.connector.connect(
+        host=parsed.hostname,
+        port=parsed.port or 3306,
+        user=parsed.username,
+        password=parsed.password,
+        database=parsed.path.lstrip("/"),
+    )
+
+
+def validate_device_type(device_type):
+    return device_type in VALID_DEVICE_TYPES
+
+
+def validate_user_key(user_key):
+    return bool(user_key) and USER_KEY_PATTERN.match(user_key)
+
+
+@app.route("/log/auth", methods=["POST"])
+@limiter.limit("30 per minute")
+def log_auth():
+    data = request.get_json(silent=True)
+    if not data:
+        return jsonify({"statusCode": 400, "message": "bad_request"}), 400
+
+    user_key = data.get("userKey", "")
+    device_type = data.get("deviceType", "")
+
+    if not validate_user_key(user_key) or not validate_device_type(device_type):
+        return jsonify({"statusCode": 400, "message": "bad_request"}), 400
+
+    try:
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO device_registrations (user_key, device_type) VALUES (%s, %s)",
+            (user_key, device_type),
+        )
+        conn.commit()
+        cursor.close()
+        conn.close()
+        return jsonify({"statusCode": 200, "message": "success"}), 200
+    except Exception:
+        return jsonify({"statusCode": 400, "message": "bad_request"}), 400
+
+
+@app.route("/log/auth/statistics", methods=["GET"])
+@limiter.limit("30 per minute")
+def log_auth_statistics():
+    device_type = request.args.get("deviceType", "")
+
+    if not validate_device_type(device_type):
+        return jsonify({"deviceType": device_type, "count": -1}), 400
+
+    try:
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT COUNT(*) FROM device_registrations WHERE device_type = %s",
+            (device_type,),
+        )
+        count = cursor.fetchone()[0]
+        cursor.close()
+        conn.close()
+        return jsonify({"deviceType": device_type, "count": count}), 200
+    except Exception:
+        return jsonify({"deviceType": device_type, "count": -1}), 500
+
+
+@app.route("/device/register", methods=["POST"])
+@limiter.limit("30 per minute")
+def register_device():
+    data = request.get_json(silent=True)
+    if not data:
+        return jsonify({"statusCode": 400}), 400
+
+    user_key = data.get("userKey", "")
+    device_type = data.get("deviceType", "")
+
+    if not validate_user_key(user_key) or not validate_device_type(device_type):
+        return jsonify({"statusCode": 400}), 400
+
+    try:
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute(
+            "INSERT INTO device_registrations (user_key, device_type) VALUES (%s, %s)",
+            (user_key, device_type),
+        )
+        conn.commit()
+        cursor.close()
+        conn.close()
+        return jsonify({"statusCode": 200}), 200
+    except Exception:
+        return jsonify({"statusCode": 400}), 400
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=5000)

docker-compose.yml

@@ -0,0 +1,32 @@
+services:
+  mariadb:
+    image: docker.io/library/mariadb:12
+    environment:
+      MARIADB_ROOT_PASSWORD: ${DB_ROOT_PASSWORD:-my-secret-pw}
+      MARIADB_DATABASE: devsecops_db
+    volumes:
+      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
+      - db_data:/var/lib/mysql
+    ports:
+      - "3306:3306"
+    healthcheck:
+      test: ["CMD", "healthcheck.sh", "--connect", "--innodb_initialized"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+
+  api:
+    image: docker.io/harshavmb/jss-devsecops-api:latest
+    build:
+      context: .
+      dockerfile: Dockerfile
+    environment:
+      DATABASE_URL: mysql://root:${DB_ROOT_PASSWORD:-my-secret-pw}@mariadb:3306/devsecops_db
+    ports:
+      - "5000:5000"
+    depends_on:
+      mariadb:
+        condition: service_healthy
+
+volumes:
+  db_data:

requirements.txt

@@ -0,0 +1,6 @@
+flask==3.1.*
+flask-limiter==3.9.*
+mysql-connector-python==9.2.*
+requests==2.32.*
+gunicorn==23.0.*
+pytest==8.3.*

tests/test_app.py

@@ -0,0 +1,164 @@
+import pytest
+import json
+from unittest.mock import patch, MagicMock
+from app.server import app
+
+
+@pytest.fixture
+def client():
+    app.config["TESTING"] = True
+    with app.test_client() as c:
+        yield c
+
+
+class TestLogAuth:
+    def test_missing_body(self, client):
+        resp = client.post("/log/auth", content_type="application/json")
+        assert resp.status_code == 400
+        assert resp.get_json()["message"] == "bad_request"
+
+    def test_invalid_device_type(self, client):
+        resp = client.post(
+            "/log/auth",
+            data=json.dumps({"userKey": "user1", "deviceType": "Nokia"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    def test_invalid_user_key(self, client):
+        resp = client.post(
+            "/log/auth",
+            data=json.dumps({"userKey": "", "deviceType": "iOS"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    def test_user_key_injection(self, client):
+        resp = client.post(
+            "/log/auth",
+            data=json.dumps({"userKey": "'; DROP TABLE--", "deviceType": "iOS"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    @patch("app.server.get_db_connection")
+    def test_success(self, mock_conn, client):
+        mock_cursor = MagicMock()
+        mock_conn.return_value.cursor.return_value = mock_cursor
+
+        resp = client.post(
+            "/log/auth",
+            data=json.dumps({"userKey": "user1", "deviceType": "iOS"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 200
+        assert resp.get_json()["message"] == "success"
+
+    @patch("app.server.get_db_connection")
+    def test_db_failure(self, mock_conn, client):
+        mock_conn.side_effect = Exception("db error")
+
+        resp = client.post(
+            "/log/auth",
+            data=json.dumps({"userKey": "user1", "deviceType": "Android"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+
+class TestLogAuthStatistics:
+    def test_invalid_device_type(self, client):
+        resp = client.get("/log/auth/statistics?deviceType=Nokia")
+        assert resp.status_code == 400
+        assert resp.get_json()["count"] == -1
+
+    def test_missing_device_type(self, client):
+        resp = client.get("/log/auth/statistics")
+        assert resp.status_code == 400
+
+    @patch("app.server.get_db_connection")
+    def test_success(self, mock_conn, client):
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = (5,)
+        mock_conn.return_value.cursor.return_value = mock_cursor
+
+        resp = client.get("/log/auth/statistics?deviceType=iOS")
+        assert resp.status_code == 200
+        data = resp.get_json()
+        assert data["deviceType"] == "iOS"
+        assert data["count"] == 5
+
+    @patch("app.server.get_db_connection")
+    def test_db_failure(self, mock_conn, client):
+        mock_conn.side_effect = Exception("connection refused")
+
+        resp = client.get("/log/auth/statistics?deviceType=Android")
+        assert resp.status_code == 500
+        assert resp.get_json()["count"] == -1
+
+
+class TestDeviceRegister:
+    def test_missing_body(self, client):
+        resp = client.post("/device/register", content_type="application/json")
+        assert resp.status_code == 400
+        assert resp.get_json()["statusCode"] == 400
+
+    def test_invalid_device_type(self, client):
+        resp = client.post(
+            "/device/register",
+            data=json.dumps({"userKey": "user1", "deviceType": "Fridge"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    def test_empty_user_key(self, client):
+        resp = client.post(
+            "/device/register",
+            data=json.dumps({"userKey": "", "deviceType": "iOS"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    def test_special_chars_rejected(self, client):
+        resp = client.post(
+            "/device/register",
+            data=json.dumps({"userKey": "<script>alert(1)</script>", "deviceType": "iOS"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    @patch("app.server.get_db_connection")
+    def test_success(self, mock_conn, client):
+        mock_cursor = MagicMock()
+        mock_conn.return_value.cursor.return_value = mock_cursor
+
+        resp = client.post(
+            "/device/register",
+            data=json.dumps({"userKey": "user1", "deviceType": "Android"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 200
+        assert resp.get_json()["statusCode"] == 200
+
+    @patch("app.server.get_db_connection")
+    def test_db_failure(self, mock_conn, client):
+        mock_conn.side_effect = Exception("db down")
+
+        resp = client.post(
+            "/device/register",
+            data=json.dumps({"userKey": "user1", "deviceType": "Watch"}),
+            content_type="application/json",
+        )
+        assert resp.status_code == 400
+
+    @patch("app.server.get_db_connection")
+    def test_all_valid_device_types(self, mock_conn, client):
+        mock_cursor = MagicMock()
+        mock_conn.return_value.cursor.return_value = mock_cursor
+        for dt in ["iOS", "Android", "Watch", "TV", "Tablet", "Desktop", "IoT"]:
+            resp = client.post(
+                "/device/register",
+                data=json.dumps({"userKey": "user1", "deviceType": dt}),
+                content_type="application/json",
+            )
+            assert resp.status_code == 200