Set Up a Canonical Academy Exam Platform – Complete Infrastructure Guide

Primary Database (Write Operations)
    ↓
Replica 1 (Read Operations)
    ↓
Replica 2 (Backup/Failover)

POST   /api/v1/auth/login
POST   /api/v1/auth/logout
GET    /api/v1/exams
GET    /api/v1/exams/{examId}
POST   /api/v1/exams/{examId}/start
POST   /api/v1/exams/{examId}/submit
GET    /api/v1/results/{resultId}

-- Users table
CREATE TABLE users (
    id UUID PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_email (email)
);
-- Exams table
CREATE TABLE exams (
    id UUID PRIMARY KEY,
    title VARCHAR(255) NOT NULL,
    description TEXT,
    duration_minutes INT NOT NULL,
    passing_score INT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Exam results table
CREATE TABLE exam_results (
    id UUID PRIMARY KEY,
    user_id UUID NOT NULL REFERENCES users(id),
    exam_id UUID NOT NULL REFERENCES exams(id),
    score INT NOT NULL,
    passed BOOLEAN NOT NULL,
    started_at TIMESTAMP NOT NULL,
    completed_at TIMESTAMP NOT NULL,
    INDEX idx_user_id (user_id),
    INDEX idx_exam_id (exam_id),
    INDEX idx_completed_at (completed_at)
);

Load Balancer (HAProxy or NGINX)
    ├── App Server 1 (Primary)
    ├── App Server 2 (Secondary)
    └── App Server 3 (Tertiary)

upstream exam_platform {
    server app1.internal:8080 weight=1;
    server app2.internal:8080 weight=1;
    server app3.internal:8080 weight=1;
    # Health check
    check interval=3000 rise=2 fall=5 timeout=1000 type=http;
    check_http_send "GET /health HTTP/1.0\r
\r
";
    check_http_expect_alive http_2xx;
}
server {
    listen 443 ssl http2;
    server_name exams.canonical.com;
    ssl_certificate /etc/ssl/certs/exam_platform.crt;
    ssl_certificate_key /etc/ssl/private/exam_platform.key;
    location / {
        proxy_pass http://exam_platform;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

# On primary server
sudo -u postgres psql -c "ALTER SYSTEM SET wal_level = replica;"
sudo -u postgres psql -c "ALTER SYSTEM SET max_wal_senders = 3;"
sudo -u postgres psql -c "ALTER SYSTEM SET max_replication_slots = 3;"
# Restart PostgreSQL
sudo systemctl restart postgresql
# Create replication user
sudo -u postgres psql -c "CREATE USER replicator WITH REPLICATION ENCRYPTED PASSWORD 'secure_password';"
# On replica server
sudo -u postgres pg_basebackup -h primary.internal -D /var/lib/postgresql/13/main -U replicator -v -P -W
# Configure recovery
echo "standby_mode = 'on'" | sudo tee -a /var/lib/postgresql/13/main/recovery.conf
echo "primary_conninfo = 'host=primary.internal port=5432 user=replicator password=secure_password'" | sudo tee -a /var/lib/postgresql/13/main/recovery.conf
sudo systemctl restart postgresql

# Install LDAP authentication module
sudo apt-get install libpam-ldapd
# Configure LDAP
sudo nano /etc/nslcd.conf
# Add:
# uri ldap://ldap.canonical.internal
# base dc=canonical,dc=com
# binddn cn=admin,dc=canonical,dc=com
# bindpw admin_password
sudo systemctl restart nslcd

# Python example using Flask and python-oauth2
from flask import Flask, redirect, url_for
from authlib.integrations.flask_client import OAuth
app = Flask(__name__)
oauth = OAuth(app)
google = oauth.register(
    name='google',
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_CLIENT_SECRET',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)
@app.route('/login')
def login():
    redirect_uri = url_for('authorize', _external=True)
    return google.authorize_redirect(redirect_uri)
@app.route('/authorize')
def authorize():
    token = google.authorize_access_token()
    user = token.get('userinfo')
    # Store user in database
    return redirect('/dashboard')

# Using pyotp for TOTP (Time-based One-Time Password)
import pyotp
import qrcode
def setup_mfa(user_id):
    secret = pyotp.random_base32()
    totp = pyotp.TOTP(secret)
    # Generate QR code
    qr = qrcode.QRCode()
    qr.add_data(totp.provisioning_uri(name=user_id, issuer_name='Canonical Academy'))
    qr.make()
    # Store secret in database
    store_mfa_secret(user_id, secret)
    return qr
def verify_mfa(user_id, token):
    secret = get_mfa_secret(user_id)
    totp = pyotp.TOTP(secret)
    return totp.verify(token)

# Exam configuration model
class ExamConfig:
    def __init__(self, exam_id):
        self.exam_id = exam_id
        self.duration_minutes = 120
        self.total_questions = 50
        self.passing_score = 70  # percentage
        self.question_randomization = True
        self.show_results_immediately = False
        self.allow_review = True
        self.allow_retakes = True
        self.retake_wait_days = 7
        self.negative_marking = False
        self.marks_per_question = 2
    def calculate_score(self, correct_answers, total_questions):
        """Calculate percentage score"""
        return (correct_answers / total_questions) * 100
    def determine_pass_fail(self, score):
        """Determine if candidate passed"""
        return score >= self.passing_score
    def get_certificate_eligible(self, score):
        """Determine if candidate qualifies for certificate"""
        return score >= 80  # Higher threshold for certificates

<!-- Proper semantic HTML -->
<form aria-label="Exam question form">
    <fieldset>
        <legend>Question 1: What is the capital of France?</legend>
        <div class="question-options">
            <label>
                <input type="radio" name="q1" value="paris" aria-describedby="q1-help">
                Paris
            </label>
            <label>
                <input type="radio" name="q1" value="london">
                London
            </label>
            <label>
                <input type="radio" name="q1" value="berlin">
                Berlin
            </label>
        </div>
        <p id="q1-help" class="help-text">Select the correct answer</p>
    </fieldset>
    <button type="submit" aria-label="Submit answer and move to next question">
        Next Question
    </button>
</form>
<style>
    /* Ensure sufficient color contrast */
    body {
        color: #212121;  /* Dark gray on white = 18:1 contrast */
        background-color: #ffffff;
    }
    /* Clear focus indicators */
    button:focus,
    input:focus {
        outline: 3px solid #4A90E2;
        outline-offset: 2px;
    }
</style>

# Install Certbot
sudo apt-get install certbot python3-certbot-nginx
# Obtain certificate
sudo certbot certonly --nginx -d exams.canonical.com -d www.exams.canonical.com
# Auto-renewal
sudo systemctl enable certbot.timer
sudo systemctl start certbot.timer

# /etc/nginx/nginx.conf
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers HIGH:!aNULL:!MD5;
ssl_prefer_server_ciphers on;
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
# HSTS (HTTP Strict Transport Security)
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;

from cryptography.fernet import Fernet
import os
# Generate encryption key (store securely in environment variables)
encryption_key = os.environ.get('ENCRYPTION_KEY')
cipher = Fernet(encryption_key)
def encrypt_sensitive_data(data):
    """Encrypt PII before storing in database"""
    return cipher.encrypt(data.encode()).decode()
def decrypt_sensitive_data(encrypted_data):
    """Decrypt PII when needed"""
    return cipher.decrypt(encrypted_data.encode()).decode()
# Usage
encrypted_ssn = encrypt_sensitive_data("123-45-6789")
# Store encrypted_ssn in database

# Enable UFW
sudo ufw enable
# Allow SSH (with rate limiting)
sudo ufw limit 22/tcp
# Allow HTTP/HTTPS
sudo ufw allow 80/tcp
sudo ufw allow 443/tcp
# Allow database replication (internal only)
sudo ufw allow from 10.0.0.0/8 to any port 5432
# Deny all other inbound traffic
sudo ufw default deny incoming
sudo ufw default allow outgoing

# Install Suricata IDS
sudo apt-get install suricata
# Configure Suricata
sudo nano /etc/suricata/suricata.yaml
# Key settings:
# - home-net: "[192.168.1.0/24,10.0.0.0/8]"
# - external-net: "!$HOME_NET"
# - threat-detection rules enabled
# Start Suricata
sudo systemctl start suricata
sudo systemctl enable suricata
# Monitor alerts
sudo tail -f /var/log/suricata/eve.json | jq 'select(.event_type=="alert")'

# Using ModSecurity with NGINX
# Install ModSecurity
sudo apt-get install libnginx-mod-http-modsecurity
# Configure ModSecurity rules
sudo nano /etc/nginx/modsecurity/modsecurity.conf
# Set: SecRuleEngine On
# Add OWASP Core Rule Set
sudo git clone https://github.com/coreruleset/coreruleset /etc/nginx/modsecurity/crs
sudo mv /etc/nginx/modsecurity/crs/crs-setup.conf.example /etc/nginx/modsecurity/crs/crs-setup.conf
# Enable in NGINX
# modsecurity on;
# modsecurity_rules_file /etc/nginx/modsecurity/main.conf;

from enum import Enum
class UserRole(Enum):
    CANDIDATE = "candidate"
    PROCTOR = "proctor"
    ADMIN = "admin"
    AUDITOR = "auditor"
class Permission(Enum):
    VIEW_OWN_RESULTS = "view_own_results"
    VIEW_ALL_RESULTS = "view_all_results"
    MANAGE_EXAMS = "manage_exams"
    MANAGE_USERS = "manage_users"
    VIEW_AUDIT_LOGS = "view_audit_logs"
# Role-permission mapping
ROLE_PERMISSIONS = {
    UserRole.CANDIDATE: [Permission.VIEW_OWN_RESULTS],
    UserRole.PROCTOR: [Permission.VIEW_ALL_RESULTS],
    UserRole.ADMIN: [Permission.MANAGE_EXAMS, Permission.MANAGE_USERS],
    UserRole.AUDITOR: [Permission.VIEW_AUDIT_LOGS]
}
def check_permission(user_role, required_permission):
    """Verify user has required permission"""
    return required_permission in ROLE_PERMISSIONS.get(user_role, [])
# Usage in API
@app.route('/api/v1/results')
def get_results():
    user = get_current_user()
    if not check_permission(user.role, Permission.VIEW_ALL_RESULTS):
        return {"error": "Unauthorized"}, 403
    return get_all_results()

from datetime import datetime, timedelta
import schedule
def delete_old_exam_data():
    """Delete exam data older than retention period"""
    retention_days = 365  # 1 year
    cutoff_date = datetime.now() - timedelta(days=retention_days)
    # Delete old results (but keep aggregate statistics)
    ExamResult.query.filter(ExamResult.completed_at < cutoff_date).delete()
    # Delete old user activity logs
    ActivityLog.query.filter(ActivityLog.timestamp < cutoff_date).delete()
    db.session.commit()
    logger.info(f"Deleted exam data older than {cutoff_date}")
# Schedule daily at 2 AM
schedule.every().day.at("02:00").do(delete_old_exam_data)

from datetime import datetime
import json
class AuditLog:
    def __init__(self, user_id, action, resource, details):
        self.user_id = user_id
        self.action = action  # CREATE, READ, UPDATE, DELETE
        self.resource = resource  # exam, result, user
        self.details = details
        self.timestamp = datetime.utcnow()
        self.ip_address = request.remote_addr
    def log(self):
        """Store audit log in database"""
        log_entry = {
            'user_id': self.user_id,
            'action': self.action,
            'resource': self.resource,
            'details': json.dumps(self.details),
            'timestamp': self.timestamp,
            'ip_address': self.ip_address
        }
        AuditLogEntry.create(**log_entry)
# Usage
def update_exam_score(exam_id, new_score):
    old_score = get_exam_score(exam_id)
    # Update score
    update_database(exam_id, new_score)
    # Log the change
    audit = AuditLog(
        user_id=current_user.id,
        action='UPDATE',
        resource='exam_result',
        details={'exam_id': exam_id, 'old_score': old_score, 'new_score': new_score}
    )
    audit.log()

# Data Processing Agreement
## Data Controller
Canonical Academy
## Data Processor
[Your Organization]
## Personal Data Processed
- Name, email, phone number
- Exam responses and scores
- IP addresses and device information
- Login timestamps
## Processing Purpose
- Exam administration and scoring
- Performance analytics
- Compliance and audit
## Data Retention
- Exam results: 3 years
- User activity logs: 1 year
- Personal data: Until account deletion + 30 days
## Security Measures
- AES-256 encryption at rest
- TLS 1.2+ encryption in transit
- Role-based access control
- Regular security audits
- Incident response procedures

# Right to access
@app.route('/api/v1/users/<user_id>/data-export', methods=['GET'])
def export_user_data(user_id):
    """Export all personal data for a user (GDPR Article 15)"""
    if not user_owns_data(user_id):
        return {"error": "Unauthorized"}, 403
    user_data = {
        'profile': get_user_profile(user_id),
        'exam_results': get_user_exam_results(user_id),
        'activity_logs': get_user_activity_logs(user_id)
    }
    return jsonify(user_data)
# Right to be forgotten
@app.route('/api/v1/users/<user_id>', methods=['DELETE'])
def delete_user_account(user_id):
    """Delete user account and associated data (GDPR Article 17)"""
    if not user_owns_data(user_id):
        return {"error": "Unauthorized"}, 403
    # Anonymize instead of delete (for audit trail)
    anonymize_user_data(user_id)
    return {"message": "Account deleted successfully"}, 200
# Consent management
class ConsentManager:
    def __init__(self, user_id):
        self.user_id = user_id
    def get_consent_status(self):
        """Get user's consent preferences"""
        return {
            'marketing_emails': get_consent(self.user_id, 'marketing'),
            'analytics': get_consent(self.user_id, 'analytics'),
            'third_party_sharing': get_consent(self.user_id, 'third_party')
        }
    def update_consent(self, consent_type, value):
        """Update user consent"""
        store_consent(self.user_id, consent_type, value)
        log_consent_change(self.user_id, consent_type, value)

# Install Prometheus
wget https://github.com/prometheus/prometheus/releases/download/v2.40.0/prometheus-2.40.0.linux-amd64.tar.gz
tar xvfz prometheus-2.40.0.linux-amd64.tar.gz
sudo mv prometheus-2.40.0.linux-amd64 /opt/prometheus
# Configure Prometheus
sudo nano /opt/prometheus/prometheus.yml

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s
scrape_configs:
  - job_name: 'exam-platform'
    static_configs:
      - targets: ['localhost:9090']
  - job_name: 'app-servers'
    static_configs:
      - targets: ['app1.internal:9100', 'app2.internal:9100', 'app3.internal:9100']
  - job_name: 'database'
    static_configs:
      - targets: ['db.internal:9187']
  - job_name: 'nginx'
    static_configs:
      - targets: ['lb.internal:9113']

# Install Grafana
sudo apt-get install grafana-server
# Start Grafana
sudo systemctl start grafana-server
sudo systemctl enable grafana-server
# Access at http://localhost:3000 (default: admin/admin)

# Install Elasticsearch
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.5.0-linux-x86_64.tar.gz
tar xvfz elasticsearch-8.5.0-linux-x86_64.tar.gz
sudo mv elasticsearch-8.5.0 /opt/elasticsearch
# Install Logstash
wget https://artifacts.elastic.co/downloads/logstash/logstash-8.5.0-linux-x86_64.tar.gz
tar xvfz logstash-8.5.0-linux-x86_64.tar.gz
sudo mv logstash-8.5.0 /opt/logstash
# Install Kibana
wget https://artifacts.elastic.co/downloads/kibana/kibana-8.5.0-linux-x86_64.tar.gz
tar xvfz kibana-8.5.0-linux-x86_64.tar.gz
sudo mv kibana-8.5.0 /opt/kibana

# /opt/logstash/config/logstash.conf
input {
  file {
    path => "/var/log/exam-platform/app.log"
    start_position => "beginning"
    codec => json
  }
  file {
    path => "/var/log/nginx/access.log"
    start_position => "beginning"
  }
  file {
    path => "/var/log/suricata/eve.json"
    start_position => "beginning"
    codec => json
  }
}
filter {
  if [type] == "app" {
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{DATA:logger} %{GREEDYDATA:message}" }
    }
  }
}
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "exam-platform-%{+YYYY.MM.dd}"
  }
}

# Alert on suspicious activities
SECURITY_ALERTS = {
    'multiple_failed_logins': {
        'condition': 'failed_login_count > 5 in 15 minutes',
        'action': 'lock_account_temporarily'
    },
    'unusual_access_pattern': {
        'condition': 'access_from_multiple_countries in 1 hour',
        'action': 'require_mfa_verification'
    },
    'data_exfiltration': {
        'condition': 'large_data_download > 100MB',
        'action': 'alert_security_team'
    },
    'privilege_escalation': {
        'condition': 'user_role_change_without_approval',
        'action': 'alert_admin'
    }
}