WP-LIT-MCP/src/Admin/ConnectionRepository.php

<?php

declare(strict_types=1);

namespace LIT\MCP\Admin;

use WPMedia\MCP\OAuth\Auth\TokenEndpoint;

/**
 * OAuth connections created by mcp-oauth (v1.1.2, TokenEndpoint): for every grant it
 * creates one application password (name = client name, app_id = the CIMD client_id,
 * always an https URL) and stores the refresh-token marker user meta
 * `mcp_refresh_jti_<uuid>`. An application password counts as an LIT MCP connection
 * only if BOTH hold: the marker meta exists and its app_id is an https URL. Any other
 * application password of the user (other tools, manual ones) is neither listed nor
 * revoked.
 *
 * Multisite: only users who are members of the current site are managed, and a
 * non-super-admin never sees or revokes a super admin's connections.
 *
 * Not final so the access tab can be tested with a mock.
 */
class ConnectionRepository
{
    public function __construct(private ?\wpdb $db = null)
    {
    }

    /**
     * @return list<array{user_id: int, user_login: string, uuid: string, name: string,
     *     app_id: string, created: int, last_used: ?int}>
     */
    public function all(): array
    {
        $wpdb = $this->db ?? $GLOBALS['wpdb'];
        $prefix = TokenEndpoint::REFRESH_JTI_META_PREFIX;
        $rows = $wpdb->get_results($wpdb->prepare(
            "SELECT user_id, meta_key FROM {$wpdb->usermeta} WHERE meta_key LIKE %s ORDER BY user_id, umeta_id",
            $wpdb->esc_like($prefix) . '%'
        ));

        $connections = [];
        foreach ((array) $rows as $row) {
            $userId = (int) $row->user_id;
            $uuid = substr((string) $row->meta_key, strlen($prefix));
            if ($uuid === '' || ! $this->manageable($userId)) {
                continue;
            }

            $item = \WP_Application_Passwords::get_user_application_password($userId, $uuid);
            if (! self::isMcpOauthPassword($item)) {
                continue;
            }

            $user = get_userdata($userId);
            $connections[] = [
                'user_id' => $userId,
                'user_login' => is_object($user) ? (string) $user->user_login : '#' . $userId,
                'uuid' => $uuid,
                'name' => (string) ($item['name'] ?? ''),
                'app_id' => (string) $item['app_id'],
                'created' => (int) ($item['created'] ?? 0),
                'last_used' => isset($item['last_used']) ? (int) $item['last_used'] : null,
            ];
        }

        return $connections;
    }

    public function revoke(int $userId, string $uuid): bool
    {
        if ($userId <= 0 || $uuid === '' || ! $this->manageable($userId)) {
            return false;
        }

        $marker = (string) get_user_meta($userId, TokenEndpoint::REFRESH_JTI_META_PREFIX . $uuid, true);
        if ($marker === '') {
            return false;
        }

        $item = \WP_Application_Passwords::get_user_application_password($userId, $uuid);
        if (! self::isMcpOauthPassword($item)) {
            return false;
        }

        // Fires wp_delete_application_password, on which mcp-oauth purges its marker meta.
        return \WP_Application_Passwords::delete_application_password($userId, $uuid) === true;
    }

    /** Whether the current administrator may see and revoke this user's connections. */
    private function manageable(int $userId): bool
    {
        if (! is_multisite()) {
            return true;
        }

        if (is_super_admin($userId) && ! is_super_admin()) {
            return false;
        }

        return is_user_member_of_blog($userId, get_current_blog_id());
    }

    private static function isMcpOauthPassword(mixed $item): bool
    {
        return is_array($item)
            && is_string($item['app_id'] ?? null)
            && str_starts_with($item['app_id'], 'https://');
    }
}