python-flask-example.py

"""
StellarGuard Python/Flask Example

Complete example of using StellarGuard SDK in a Flask application
"""

import os
from functools import wraps
from typing import Optional, List

from flask import Flask, request, jsonify, render_template_string
from stellar_guard import (
    StellarGuard,
    Permission,
    StellarGuardConfig,
    RoleBasedPolicy,
    AuthorizationError,
    RoleNotFoundError,
    StellarGuardError,
)

# ============================================================================
# Configuration
# ============================================================================

app = Flask(__name__)

config = StellarGuardConfig(
    contract_id=os.getenv("CONTRACT_ID", ""),
    rpc_url=os.getenv("RPC_URL", "https://soroban-testnet.stellar.org"),
    network_passphrase=os.getenv(
        "NETWORK_PASSPHRASE", "Test SDF Network ; September 2015"
    ),
)

sg = StellarGuard(config)

# ============================================================================
# Middleware & Decorators
# ============================================================================


def get_user_address():
    """Extract user address from request header."""
    user_address = request.headers.get("X-User-Address")
    if not user_address:
        raise ValueError("Missing X-User-Address header")
    return user_address


def require_auth(func):
    """Decorator: Require authentication."""

    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            user_address = get_user_address()
            return func(user_address, *args, **kwargs)
        except ValueError as e:
            return jsonify({"error": str(e)}), 401
        except Exception as e:
            return jsonify({"error": str(e)}), 500

    return wrapper


def require_permission(permission: Permission):
    """Decorator: Require specific permission."""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                user_address = get_user_address()

                # Check permission
                if not sg.has_permission(user_address, permission):
                    return jsonify({"error": "Forbidden"}), 403

                return func(user_address, *args, **kwargs)
            except ValueError as e:
                return jsonify({"error": str(e)}), 401
            except StellarGuardError as e:
                return jsonify({"error": str(e)}), 500
            except Exception as e:
                return jsonify({"error": str(e)}), 500

        return wrapper

    return decorator


def require_role(role_name: str):
    """Decorator: Require specific role."""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                user_address = get_user_address()

                # Check role
                if not sg.has_role(user_address, role_name):
                    return jsonify({"error": "Forbidden"}), 403

                return func(user_address, *args, **kwargs)
            except ValueError as e:
                return jsonify({"error": str(e)}), 401
            except StellarGuardError as e:
                return jsonify({"error": str(e)}), 500
            except Exception as e:
                return jsonify({"error": str(e)}), 500

        return wrapper

    return decorator


# ============================================================================
# API Routes - User Management
# ============================================================================


@app.route("/api/user/profile", methods=["GET"])
@require_auth
def get_user_profile(user_address: str):
    """Get user profile with roles and permissions."""
    try:
        roles = sg.get_member_roles(user_address)
        access_info = sg.verify_member_access(user_address)

        return jsonify(
            {
                "user": user_address,
                "roles": roles,
                "permissions": [p.value if hasattr(p, "value") else str(p) for p in access_info.permissions],
                "canAccess": access_info.can_access,
            }
        )
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/user/roles", methods=["GET"])
@require_auth
def get_user_roles(user_address: str):
    """Get user's roles."""
    try:
        roles = sg.get_member_roles(user_address)
        return jsonify({"roles": roles})
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


# ============================================================================
# API Routes - Data Operations
# ============================================================================


@app.route("/api/data", methods=["GET"])
@require_permission(Permission.READ)
def get_data(user_address: str):
    """Get data (requires read permission)."""
    return jsonify(
        {"data": "This is sensitive information", "user": user_address}
    )


@app.route("/api/data", methods=["POST"])
@require_permission(Permission.WRITE)
def create_data(user_address: str):
    """Create data (requires write permission)."""
    data = request.get_json()
    return jsonify(
        {"status": "created", "data": data, "createdBy": user_address}
    )


@app.route("/api/data/<item_id>", methods=["DELETE"])
@require_permission(Permission.DELETE)
def delete_data(user_address: str, item_id: str):
    """Delete data (requires delete permission)."""
    return jsonify(
        {"status": "deleted", "itemId": item_id, "deletedBy": user_address}
    )


# ============================================================================
# API Routes - Admin Operations
# ============================================================================


@app.route("/api/admin/roles", methods=["GET"])
@require_role("admin")
def list_roles(user_address: str):
    """List all roles (admin only)."""
    try:
        limit = request.args.get("limit", 50, type=int)
        offset = request.args.get("offset", 0, type=int)

        roles_page = sg.list_roles(limit=limit, offset=offset)

        return jsonify(
            {
                "data": roles_page.data,
                "limit": roles_page.limit,
                "offset": roles_page.offset,
                "total": roles_page.total,
                "hasMore": roles_page.has_more,
            }
        )
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/admin/roles", methods=["POST"])
@require_role("admin")
def create_role(user_address: str):
    """Create new role (admin only)."""
    try:
        data = request.get_json()

        result = sg.create_role(
            admin=user_address,
            role_name=data["name"],
            permissions=[Permission[p.upper()] for p in data.get("permissions", [])],
            description=data.get("description"),
        )

        return jsonify(
            {"status": "created", "txHash": result.tx_hash}
        ), 201
    except RoleNotFoundError as e:
        return jsonify({"error": str(e)}), 404
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/admin/members/<member>/roles/<role>", methods=["POST"])
@require_role("admin")
def grant_role(user_address: str, member: str, role: str):
    """Grant role to member (admin only)."""
    try:
        result = sg.grant_role(admin=user_address, member=member, role_name=role)
        return jsonify(
            {"status": "granted", "txHash": result.tx_hash}
        ), 200
    except RoleNotFoundError as e:
        return jsonify({"error": str(e)}), 404
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/admin/members/<member>/roles/<role>", methods=["DELETE"])
@require_role("admin")
def revoke_role(user_address: str, member: str, role: str):
    """Revoke role from member (admin only)."""
    try:
        result = sg.revoke_role(admin=user_address, member=member, role_name=role)
        return jsonify(
            {"status": "revoked", "txHash": result.tx_hash}
        ), 200
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


# ============================================================================
# API Routes - Statistics
# ============================================================================


@app.route("/api/stats", methods=["GET"])
@require_auth
def get_stats(user_address: str):
    """Get contract statistics."""
    try:
        stats = sg.get_contract_stats()

        return jsonify(
            {
                "totalRoles": stats.total_roles,
                "totalMembers": stats.total_members,
                "totalPermissions": stats.total_permissions,
                "lastUpdated": stats.last_updated,
            }
        )
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/admin/role/<role>/members", methods=["GET"])
@require_role("admin")
def get_role_members(user_address: str, role: str):
    """Get members with specific role (admin only)."""
    try:
        limit = request.args.get("limit", 50, type=int)
        offset = request.args.get("offset", 0, type=int)

        members_page = sg.get_role_members(role, limit=limit, offset=offset)

        return jsonify(
            {
                "data": members_page.data,
                "limit": members_page.limit,
                "offset": members_page.offset,
                "total": members_page.total,
                "hasMore": members_page.has_more,
            }
        )
    except StellarGuardError as e:
        return jsonify({"error": str(e)}), 500


# ============================================================================
# Health Check
# ============================================================================


@app.route("/health", methods=["GET"])
def health_check():
    """Health check endpoint."""
    return jsonify({"status": "ok"}), 200


# ============================================================================
# Error Handlers
# ============================================================================


@app.errorhandler(404)
def not_found(error):
    """Handle 404 errors."""
    return jsonify({"error": "Not found"}), 404


@app.errorhandler(500)
def internal_error(error):
    """Handle 500 errors."""
    return jsonify({"error": "Internal server error"}), 500


# ============================================================================
# Simple HTML Dashboard
# ============================================================================

HTML_DASHBOARD = """
<!DOCTYPE html>
<html>
<head>
    <title>StellarGuard Dashboard</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        .container { max-width: 1200px; margin: 0 auto; }
        .card { border: 1px solid #ddd; border-radius: 5px; padding: 15px; margin-bottom: 15px; }
        .stats { display: grid; grid-template-columns: repeat(3, 1fr); gap: 15px; }
        .stat-item { background: #f0f0f0; padding: 15px; border-radius: 5px; text-align: center; }
        .stat-number { font-size: 32px; font-weight: bold; }
        button { padding: 10px 20px; background: #007bff; color: white; border: none; border-radius: 5px; cursor: pointer; }
        button:hover { background: #0056b3; }
    </style>
</head>
<body>
    <div class="container">
        <h1>StellarGuard Dashboard</h1>

        <div class="card">
            <h2>Set User Address</h2>
            <input type="text" id="userAddress" placeholder="Enter your Stellar address" style="width: 100%; padding: 10px;">
            <button onclick="setUserAddress()">Set</button>
        </div>

        <div class="card">
            <h2>Statistics</h2>
            <div class="stats" id="statsContainer">
                <div class="stat-item">
                    <div class="stat-number" id="totalRoles">-</div>
                    <div>Total Roles</div>
                </div>
                <div class="stat-item">
                    <div class="stat-number" id="totalMembers">-</div>
                    <div>Total Members</div>
                </div>
                <div class="stat-item">
                    <div class="stat-number" id="totalPermissions">-</div>
                    <div>Total Permissions</div>
                </div>
            </div>
        </div>

        <div class="card">
            <h2>User Roles</h2>
            <div id="userRoles">
                <p>Set user address to see roles</p>
            </div>
        </div>
    </div>

    <script>
        let currentUserAddress = localStorage.getItem("userAddress") || "";

        function setUserAddress() {
            currentUserAddress = document.getElementById("userAddress").value;
            localStorage.setItem("userAddress", currentUserAddress);
            loadData();
        }

        function loadData() {
            if (!currentUserAddress) return;

            fetch("/api/stats", {
                headers: { "X-User-Address": currentUserAddress }
            })
            .then(r => r.json())
            .then(data => {
                document.getElementById("totalRoles").textContent = data.totalRoles;
                document.getElementById("totalMembers").textContent = data.totalMembers;
                document.getElementById("totalPermissions").textContent = data.totalPermissions;
            });

            fetch("/api/user/roles", {
                headers: { "X-User-Address": currentUserAddress }
            })
            .then(r => r.json())
            .then(data => {
                document.getElementById("userRoles").innerHTML =
                    data.roles.length > 0
                        ? "<ul>" + data.roles.map(r => "<li>" + r + "</li>").join("") + "</ul>"
                        : "<p>No roles assigned</p>";
            });
        }

        if (currentUserAddress) {
            document.getElementById("userAddress").value = currentUserAddress;
            loadData();
        }

        // Auto-refresh every 30 seconds
        setInterval(loadData, 30000);
    </script>
</body>
</html>
"""


@app.route("/")
def dashboard():
    """Serve HTML dashboard."""
    return render_template_string(HTML_DASHBOARD)


# ============================================================================
# Main
# ============================================================================

if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=5000)