api_server.py

#!/usr/bin/env python3
"""
SQLite-powered Mock API Server for DOE Hardware Integration
Serves Intune and DeviceHub APIs from SQLite database
"""

from flask import Flask, request, jsonify
import sqlite3
import json
import uuid
from datetime import datetime

app = Flask(__name__)

# Database configuration
DB_FILE = "doe_devices.db"

def get_db_connection():
    """Get SQLite database connection"""
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row  # Return rows as dictionaries
    return conn

def format_intune_device(row):
    """Convert database row to Intune API format"""
    return {
        "id": row['id'],
        "userId": str(uuid.uuid4()),
        "deviceName": row['device_name'],
        "ownerType": "company",
        "managedDeviceOwnerType": "company",
        "managementState": row['management_state'],
        "enrolledDateTime": row['enrolled_date_time'],
        "lastSyncDateTime": row['last_sync_date_time'],
        "complianceState": row['compliance_state'],
        "jailBroken": "Unknown",
        "managementAgent": "mdm",
        "osVersion": row['os_version'],
        "easActivated": True,
        "easDeviceId": str(uuid.uuid4()),
        "easActivationDateTime": row['enrolled_date_time'],
        "aadRegistered": True,
        "azureADRegistered": True,
        "deviceEnrollmentType": "windowsAzureADJoin",
        "lostModeState": "disabled",
        "activationLockBypassCode": None,
        "emailAddress": row['user_principal_name'],
        "azureActiveDirectoryDeviceId": str(uuid.uuid4()),
        "azureADDeviceId": str(uuid.uuid4()),
        "deviceRegistrationState": "registered",
        "deviceCategoryDisplayName": row['device_category'],
        "isSupervised": False,
        "exchangeLastSuccessfulSyncDateTime": row['last_sync_date_time'],
        "exchangeAccessState": "allowed",
        "exchangeAccessStateReason": "none",
        "remoteAssistanceSessionUrl": None,
        "remoteAssistanceSessionErrorDetails": None,
        "isEncrypted": True,
        "userPrincipalName": row['user_principal_name'],
        "model": row['model'],
        "manufacturer": row['manufacturer'],
        "imei": None,
        "complianceGracePeriodExpirationDateTime": "2025-10-09T18:44:38.127Z",
        "serialNumber": row['serial_number'],
        "phoneNumber": None,
        "androidSecurityPatchLevel": None,
        "userDisplayName": "NYC DOE User",
        "configurationManagerClientEnabledFeatures": None,
        "wiFiMacAddress": row['wifi_mac_address'],
        "deviceHealthAttestationState": None,
        "subscriberCarrier": None,
        "meid": None,
        "totalStorageSpaceInBytes": row['total_storage_bytes'],
        "freeStorageSpaceInBytes": int(row['total_storage_bytes'] * 0.6),  # Mock free space
        "managedDeviceName": row['device_name'],
        "partnerReportedThreatState": "activated",
        "retireAfterDateTime": "2028-09-08T18:44:38.127Z",
        "usersLoggedOn": [],
        "preferMdmOverGroupPolicyAppliedDateTime": row['enrolled_date_time'],
        "autopilotEnrolled": True,
        "requireUserEnrollmentApproval": False,
        "managementCertificateExpirationDate": "2026-09-09T18:44:38.127Z",
        "iccid": None,
        "udid": None,
        "roleScopeTagIds": ["0"],
        "windowsActiveMalwareCount": 0,
        "windowsRemediatedMalwareCount": 0,
        "notes": row['notes'],
        "configurationManagerClientHealthState": None,
        "configurationManagerClientInformation": None,
        "ethernetMacAddress": row['wifi_mac_address'].replace(':', '').upper(),
        "physicalMemoryInBytes": row['physical_memory_bytes'],
        "processorArchitecture": "x64",
        "specificationVersion": None,
        "joinType": "azureADJoined",
        "skuFamily": None,
        "skuNumber": None,
        "managementFeatures": "none",
        "chromeOSDeviceInfo": [{
            "annotatedAssetId": row['serial_number'].replace('NYC-SN', 'NYC-AT'),
            "annotatedLocation": "NYC Department of Education",
            "annotatedUser": row['user_principal_name'],
            "lastSyncDateTime": row['last_sync_date_time'],
            "supportEndDateTime": "2029-09-03T00:00:00.000Z"
        }],
        "operatingSystem": row['operating_system'],
        "deviceType": "chromeOS"
    }

def format_devicehub_asset(row):
    """Convert database row to DeviceHub API format"""
    return {
        "assetId": row['asset_id'],
        "assetTag": row['asset_tag'],
        "manufacturer": row['manufacturer'],
        "model": row['model'],
        "category": row['category'],
        "status": row['status'],
        "location": {
            "building": row['location_building']
        } if row['location_building'] else None,
        "assignedTo": {
            "name": row['assigned_to_name'],
            "department": row['assigned_to_department']
        } if row['assigned_to_name'] else None,
        "purchaseInfo": {
            "purchasePrice": row['purchase_price'],
            "purchaseDate": row['purchase_date'] if row['purchase_date'] else "",
            "warrantyExpiration": row['warranty_expiration']
        }
    }

@app.route('/api/intune/deviceManagement/managedDevices', methods=['GET'])
def get_intune_devices():
    """Get Intune devices with optional filtering"""
    conn = get_db_connection()
    cursor = conn.cursor()
    
    # Parse filter parameter
    filter_param = request.args.get('$filter', '')
    
    if filter_param:
        # Look for serial number in filter (simple parsing)
        serial_number = filter_param.strip()
        cursor.execute('''
            SELECT * FROM intune_devices 
            WHERE serial_number = ?
        ''', (serial_number,))
    else:
        # Return all devices (with pagination support)
        limit = int(request.args.get('$top', 100))  # Default 100 items
        offset = int(request.args.get('$skip', 0))
        cursor.execute('''
            SELECT * FROM intune_devices 
            ORDER BY serial_number
            LIMIT ? OFFSET ?
        ''', (limit, offset))
    
    devices = cursor.fetchall()
    conn.close()
    
    # Format response
    response = {
        "@odata.context": "https://graph.microsoft.com/v1.0/$metadata#deviceManagement/managedDevices",
        "value": [format_intune_device(device) for device in devices]
    }
    
    return jsonify(response)

@app.route('/api/devicehub/assets/serial/<serial_number>', methods=['GET'])
def get_devicehub_asset_by_serial(serial_number):
    """Get DeviceHub asset by serial number"""
    conn = get_db_connection()
    cursor = conn.cursor()
    
    cursor.execute('''
        SELECT * FROM devicehub_assets 
        WHERE serial_number = ?
    ''', (serial_number,))
    
    asset = cursor.fetchone()
    conn.close()
    
    if asset:
        return jsonify(format_devicehub_asset(asset))
    else:
        return jsonify({"error": "Asset not found"}), 404

@app.route('/api/devicehub/assets', methods=['GET'])
def get_devicehub_assets():
    """Get all DeviceHub assets with pagination"""
    conn = get_db_connection()
    cursor = conn.cursor()
    
    limit = int(request.args.get('limit', 100))
    offset = int(request.args.get('offset', 0))
    
    cursor.execute('''
        SELECT * FROM devicehub_assets 
        ORDER BY serial_number
        LIMIT ? OFFSET ?
    ''', (limit, offset))
    
    assets = cursor.fetchall()
    conn.close()
    
    response = {
        "assets": [format_devicehub_asset(asset) for asset in assets],
        "pagination": {
            "limit": limit,
            "offset": offset,
            "count": len(assets)
        }
    }
    
    return jsonify(response)

@app.route('/health', methods=['GET'])
def health_check():
    """Health check endpoint"""
    conn = get_db_connection()
    cursor = conn.cursor()
    
    # Check database connectivity and count records
    cursor.execute("SELECT COUNT(*) as device_count FROM devices")
    device_count = cursor.fetchone()['device_count']
    
    cursor.execute("SELECT COUNT(*) as intune_count FROM intune_devices")
    intune_count = cursor.fetchone()['intune_count']
    
    cursor.execute("SELECT COUNT(*) as devicehub_count FROM devicehub_assets")
    devicehub_count = cursor.fetchone()['devicehub_count']
    
    conn.close()
    
    return jsonify({
        "status": "healthy",
        "database": DB_FILE,
        "records": {
            "devices": device_count,
            "intune": intune_count,
            "devicehub": devicehub_count
        },
        "endpoints": {
            "intune": "/api/intune/deviceManagement/managedDevices",
            "devicehub": "/api/devicehub/assets"
        }
    })

@app.route('/', methods=['GET'])
def index():
    """API documentation"""
    return jsonify({
        "name": "DOE Mock APIs - SQLite Edition",
        "description": "Mock Intune and DeviceHub APIs powered by SQLite database",
        "endpoints": {
            "/health": "Health check and database statistics",
            "/api/intune/deviceManagement/managedDevices": "Intune devices API",
            "/api/intune/deviceManagement/managedDevices?$filter=SERIAL": "Filter Intune by serial",
            "/api/devicehub/assets": "DeviceHub assets API",
            "/api/devicehub/assets/serial/{serial}": "DeviceHub asset by serial"
        },
        "database": {
            "file": DB_FILE,
            "total_devices": "~350K from NYC CSV"
        }
    })

if __name__ == '__main__':
    print("🚀 Starting DOE Mock API Server (SQLite Edition)")
    print(f"📊 Database: {DB_FILE}")
    print("🌐 Endpoints:")
    print("   - http://localhost:5000/ (documentation)")
    print("   - http://localhost:5000/health (health check)")
    print("   - http://localhost:5000/api/intune/deviceManagement/managedDevices")
    print("   - http://localhost:5000/api/devicehub/assets")
    print()
    
    app.run(debug=True, host='0.0.0.0', port=5000)