Incorrect Use of Privileged APIs

Description

Incorrect Use of Privileged APIs occurs when an application fails to properly control access to or invocation of privileged system functions. This includes calling APIs that require elevated privileges without proper authorization checks, exposing privileged APIs to unprivileged users, or failing to drop privileges after completing privileged operations. The vulnerability allows attackers to perform operations they should not be authorized to execute.

Risk

Unprivileged users can execute administrative functions. Privilege escalation through improperly exposed APIs. System commands executed with unnecessary elevated privileges. Sensitive operations performed without authorization. Security boundaries violated through API abuse. Full system compromise through privileged API access.

Solution

Implement proper authorization before calling privileged APIs. Follow principle of least privilege. Drop privileges immediately after privileged operations. Use capability-based security where available. Audit privileged API usage. Separate privileged and unprivileged code paths. Use security frameworks for access control.

Common Consequences

Impact	Details
Authorization	Scope: Privilege Escalation Users gain unauthorized elevated access.
Integrity	Scope: System Modification Unauthorized changes to system state.
Confidentiality	Scope: Data Access Access to privileged data through API abuse.

Example Code + Solution Code

Vulnerable Code

// VULNERABLE: Privileged API exposed without authorization
@RestController
@RequestMapping("/api")
public class VulnerableSystemController {

    // VULNERABLE: Anyone can call system commands
    @PostMapping("/exec")
    public String executeCommand(@RequestBody String command) {
        // No authorization check!
        Runtime runtime = Runtime.getRuntime();
        Process process = runtime.exec(command);
        return readOutput(process);
    }

    // VULNERABLE: File operations with elevated privileges
    @PostMapping("/system/write")
    public void writeSystemFile(@RequestBody FileRequest request) {
        // No authorization - anyone can write anywhere!
        Files.write(Paths.get(request.getPath()), request.getContent().getBytes());
    }

    // VULNERABLE: User management without authorization
    @PostMapping("/users/create")
    public void createUser(@RequestBody User user) {
        // Should require admin, but no check!
        userService.createUser(user);
    }
}

// VULNERABLE: Privileges not dropped after use
public class VulnerablePrivilegedOperation {

    public void performOperation() {
        // Elevate privileges
        AccessController.doPrivileged(new PrivilegedAction<Void>() {
            public Void run() {
                // Perform privileged operation
                readSensitiveConfig();

                // VULNERABLE: Continue running privileged
                // even for non-privileged operations
                handleUserInput();  // Should NOT be privileged!

                return null;
            }
        });
    }
}

# VULNERABLE: Privileged operations without checks
import os
import subprocess
from flask import Flask, request

app = Flask(__name__)

@app.route('/api/admin/exec', methods=['POST'])
def exec_command_vulnerable():
    # VULNERABLE: No authorization check
    command = request.json['command']
    result = subprocess.run(command, shell=True, capture_output=True)
    return {'output': result.stdout.decode()}

@app.route('/api/system/config', methods=['POST'])
def write_config_vulnerable():
    # VULNERABLE: Anyone can write system config
    config_path = request.json['path']
    content = request.json['content']

    # Writing to system paths without authorization!
    with open(config_path, 'w') as f:
        f.write(content)

    return {'status': 'written'}

# VULNERABLE: Running entire app as root
# The application runs with root privileges for everything
# even operations that don't need it

@app.route('/api/user/profile')
def get_profile_vulnerable():
    # This simple operation runs with full root privileges!
    user_id = request.args.get('id')
    return get_user_profile(user_id)

# VULNERABLE: setuid not properly managed
def perform_privileged_task():
    # Elevate to root
    os.seteuid(0)

    # Perform privileged operation
    perform_admin_task()

    # VULNERABLE: Forgot to drop privileges!
    # All subsequent code runs as root

    handle_user_request()  # Running as root unnecessarily

// VULNERABLE: Node.js privileged API exposure
const express = require('express');
const { exec } = require('child_process');
const fs = require('fs');

const app = express();

// VULNERABLE: Anyone can execute system commands
app.post('/api/system/exec', (req, res) => {
    const { command } = req.body;
    // No authorization check!
    exec(command, (error, stdout) => {
        res.json({ output: stdout });
    });
});

// VULNERABLE: Admin API without authorization
app.post('/api/admin/users', (req, res) => {
    // Should require admin role but doesn't check
    const userData = req.body;
    db.createUser(userData);  // Creates admin users!
    res.json({ success: true });
});

// VULNERABLE: Filesystem operations exposed
app.post('/api/files/write', (req, res) => {
    const { path, content } = req.body;
    // No check - anyone can write anywhere!
    fs.writeFileSync(path, content);
    res.json({ success: true });
});

// VULNERABLE: Database admin operations
app.post('/api/db/execute', (req, res) => {
    const { query } = req.body;
    // No authorization - raw SQL execution!
    db.raw(query).then(result => {
        res.json(result);
    });
});

// VULNERABLE: C program with improper privilege handling
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>

// VULNERABLE: Running as root unnecessarily
void vulnerable_process() {
    // Entire function runs as root
    // even for unprivileged operations

    read_user_input();  // Doesn't need root!
    log_activity();     // Doesn't need root!

    // Only this needs root
    write_system_file("/etc/config", data);

    // Still running as root after!
    process_user_request();  // Dangerous!
}

// VULNERABLE: setuid not dropped
void vulnerable_setuid_program() {
    // Program is setuid root

    // VULNERABLE: Never drops privileges
    while (1) {
        char* command = read_user_command();
        // Executes user commands as root!
        system(command);
    }
}

// VULNERABLE: Privilege check bypass
int check_admin() {
    // VULNERABLE: Can be bypassed
    return getenv("IS_ADMIN") != NULL;
}

void admin_function() {
    // Attacker sets IS_ADMIN=1 environment variable
    if (check_admin()) {
        // Performs privileged operations
        delete_all_users();
    }
}

Fixed Code

// SAFE: Proper authorization for privileged APIs
@RestController
@RequestMapping("/api")
public class SafeSystemController {

    private final AuthorizationService authService;

    // SAFE: Strict authorization for command execution
    @PreAuthorize("hasRole('SYSTEM_ADMIN')")
    @PostMapping("/exec")
    public String executeCommand(@AuthenticationPrincipal User user,
                                 @RequestBody String command) {
        // Validate user has required role
        if (!authService.canExecuteSystemCommands(user)) {
            throw new AccessDeniedException("Not authorized");
        }

        // Whitelist allowed commands
        if (!isAllowedCommand(command)) {
            throw new IllegalArgumentException("Command not allowed");
        }

        // Audit log
        auditService.logPrivilegedOperation(user, "exec", command);

        // Execute with minimal privileges
        return executeWithLimitedPrivileges(command);
    }

    private boolean isAllowedCommand(String command) {
        Set<String> allowed = Set.of("status", "health", "metrics");
        return allowed.contains(command);
    }
}

// SAFE: Minimal privilege scope
public class SafePrivilegedOperation {

    public void performOperation(User user) {
        // Check authorization first
        if (!user.hasRole("ADMIN")) {
            throw new AccessDeniedException("Admin required");
        }

        // Only elevate for specific operation
        String config = AccessController.doPrivileged(
            (PrivilegedAction<String>) this::readSensitiveConfig
        );
        // Privileges automatically dropped here

        // Continue with normal privileges
        processConfig(config);  // Not privileged
    }

    // SAFE: Separate privileged and unprivileged methods
    @Privileged
    private String readSensitiveConfig() {
        return Files.readString(Path.of("/etc/app/config"));
    }

    private void processConfig(String config) {
        // Runs with normal privileges
    }
}

# SAFE: Proper privilege management
import os
from functools import wraps
from flask import Flask, request, g

app = Flask(__name__)

def require_role(role):
    """Decorator to require specific role for endpoint."""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            if not g.current_user:
                return {'error': 'Not authenticated'}, 401
            if not g.current_user.has_role(role):
                return {'error': 'Insufficient privileges'}, 403
            return f(*args, **kwargs)
        return wrapper
    return decorator

def audit_privileged_action(action):
    """Log privileged actions for audit trail."""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            audit_log.log(
                user=g.current_user,
                action=action,
                params=request.json
            )
            return f(*args, **kwargs)
        return wrapper
    return decorator

@app.route('/api/admin/exec', methods=['POST'])
@require_role('system_admin')
@audit_privileged_action('command_execution')
def exec_command_safe():
    command = request.json['command']

    # Whitelist allowed commands
    allowed_commands = {'status', 'health', 'restart'}
    if command not in allowed_commands:
        return {'error': 'Command not allowed'}, 400

    # Execute with dropped privileges
    result = run_as_unprivileged_user(command)
    return {'output': result}

# SAFE: Proper setuid handling
def perform_privileged_task_safe(user):
    # Verify authorization
    if not user.is_admin:
        raise PermissionError("Admin required")

    # Save original uid
    original_euid = os.geteuid()

    try:
        # Elevate only for specific operation
        os.seteuid(0)
        perform_admin_task()
    finally:
        # Always drop privileges
        os.seteuid(original_euid)

    # Continue with normal privileges
    handle_user_request()

# SAFE: Context manager for privilege elevation
class PrivilegeElevation:
    def __init__(self, required_role):
        self.required_role = required_role
        self.original_euid = None

    def __enter__(self):
        # Check authorization before elevating
        if not current_user.has_role(self.required_role):
            raise PermissionError(f"Role {self.required_role} required")

        self.original_euid = os.geteuid()
        os.seteuid(0)
        return self

    def __exit__(self, *args):
        # Always restore original privileges
        if self.original_euid is not None:
            os.seteuid(self.original_euid)

# Usage
with PrivilegeElevation('admin'):
    write_system_config()
# Privileges dropped automatically

// SAFE: Node.js with proper authorization
const express = require('express');
const app = express();

// Authorization middleware
function requireRole(role) {
    return (req, res, next) => {
        if (!req.user) {
            return res.status(401).json({ error: 'Not authenticated' });
        }
        if (!req.user.roles.includes(role)) {
            return res.status(403).json({ error: 'Insufficient privileges' });
        }
        next();
    };
}

// Audit logging
function auditLog(action) {
    return (req, res, next) => {
        logger.info({
            action,
            user: req.user.id,
            ip: req.ip,
            params: req.body
        });
        next();
    };
}

// SAFE: Protected privileged endpoint
app.post('/api/admin/users',
    requireRole('admin'),
    auditLog('create_user'),
    async (req, res) => {
        const userData = req.body;

        // Validate request
        if (!validateUserData(userData)) {
            return res.status(400).json({ error: 'Invalid data' });
        }

        // Only allow creating non-admin users
        if (userData.roles?.includes('admin')) {
            if (!req.user.roles.includes('super_admin')) {
                return res.status(403).json({
                    error: 'Cannot create admin users'
                });
            }
        }

        await db.createUser(userData);
        res.json({ success: true });
    }
);

// SAFE: Whitelist allowed operations
const ALLOWED_OPERATIONS = new Set(['status', 'health', 'restart']);

app.post('/api/system/operation',
    requireRole('system_admin'),
    auditLog('system_operation'),
    (req, res) => {
        const { operation } = req.body;

        if (!ALLOWED_OPERATIONS.has(operation)) {
            return res.status(400).json({ error: 'Operation not allowed' });
        }

        // Execute allowed operation
        const result = executeSystemOperation(operation);
        res.json(result);
    }
);

// SAFE: C program with proper privilege dropping
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <pwd.h>
#include <grp.h>

// SAFE: Drop privileges immediately after use
void safe_privileged_operation() {
    uid_t original_uid = getuid();
    uid_t original_euid = geteuid();

    // Only the privileged operation runs as root
    if (geteuid() == 0) {
        write_system_file("/etc/config", data);
    }

    // Drop privileges permanently
    if (drop_privileges() != 0) {
        fprintf(stderr, "Failed to drop privileges\n");
        exit(1);
    }

    // All subsequent code runs unprivileged
    read_user_input();  // Safe - not root
    process_request();  // Safe - not root
}

int drop_privileges() {
    struct passwd *pw = getpwnam("nobody");
    if (pw == NULL) {
        return -1;
    }

    // Drop supplementary groups
    if (setgroups(0, NULL) != 0) {
        return -1;
    }

    // Drop to unprivileged user
    if (setgid(pw->pw_gid) != 0) {
        return -1;
    }
    if (setuid(pw->pw_uid) != 0) {
        return -1;
    }

    // Verify privileges were dropped
    if (getuid() == 0 || geteuid() == 0) {
        return -1;  // Still root - something wrong
    }

    return 0;
}

// SAFE: Separate privileged operations
void safe_setuid_program() {
    // Drop privileges immediately at startup
    uid_t real_uid = getuid();

    if (geteuid() == 0) {
        // Only perform specific privileged operations
        bind_privileged_port(80);
        read_ssl_keys();

        // Drop privileges permanently
        if (drop_privileges() != 0) {
            exit(1);
        }
    }

    // Main loop runs unprivileged
    while (1) {
        handle_request();  // Safe - not root
    }
}

---

Exploited in the Wild

Container Escapes

Privileged API access leading to container breakout.

Admin API Exposure

Unauthenticated admin API endpoints.

Privilege Retention

Services retaining elevated privileges unnecessarily.

---

Tools to test/exploit

• API security scanners.

• Privilege escalation testing tools.

• Access control testing.

---

CVE Examples

• CVE-2019-5736: Container runtime privilege escalation.

• Numerous API authorization CVEs.

---

References

1. MITRE. "CWE-648: Incorrect Use of Privileged APIs." https://cwe.mitre.org/data/definitions/648.html

2. OWASP. "Principle of Least Privilege."