diff --git a/doc/internal/MySQL_AuthPlugin.md b/doc/internal/MySQL_AuthPlugin.md
new file mode 100644
index 0000000000..12139adc6c
--- /dev/null
+++ b/doc/internal/MySQL_AuthPlugin.md
@@ -0,0 +1,334 @@
+# Per-User Authentication Plugins Design
+
+## Overview
+
+This document describes the per-user authentication plugin framework in ProxySQL. Users can authenticate using different methods (e.g., Kubernetes ServiceAccount tokens, static passwords) by setting the `auth_plugin` attribute in `mysql_users.attributes`.
+
+## Motivation
+
+ProxySQL already supports:
+1. **Client/backend auth separation**: Clients can authenticate to ProxySQL differently than ProxySQL authenticates to backends
+2. **LDAP authentication**: Via an enterprise plugin loaded at startup
+3. **User attributes**: JSON field in `mysql_users` for per-user configuration
+
+This framework extends the architecture to support per-user auth plugin selection for users in `mysql_users`, enabling scenarios like Kubernetes ServiceAccount authentication.
+
+### Use Case: Kubernetes ServiceAccount Authentication
+
+Applications running in Kubernetes authenticate to ProxySQL using their ServiceAccount token:
+
+```
+Client (SA token) → ProxySQL (validates via TokenReview API) → MariaDB (regular credentials)
+```
+
+Benefits:
+- No static passwords in application configs
+- Tokens are short-lived and auto-rotated
+- Identity tied to Kubernetes workload identity
+
+## User Configuration
+
+### Enabling Auth Plugin for a User
+
+```sql
+-- User with static auth plugin (frontend-only)
+INSERT INTO mysql_users (username, password, attributes, default_hostgroup, frontend, backend)
+VALUES (
+    'app_user',
+    '',
+    '{"auth_plugin": "static", "static_password": "secret123", "backend_username": "dbuser"}',
+    1, 1, 0
+);
+
+-- Backend-only user for the actual database connection
+INSERT INTO mysql_users (username, password, default_hostgroup, frontend, backend)
+VALUES ('dbuser', 'db_password', 1, 0, 1);
+
+LOAD MYSQL USERS TO RUNTIME;
+```
+
+### Standard Users (Unchanged)
+
+```sql
+-- Users without auth_plugin attribute use standard password auth
+INSERT INTO mysql_users (username, password, default_hostgroup)
+VALUES ('regular_user', 'password123', 1);
+```
+
+## ProxySQL Configuration
+
+### Config File (`proxysql.cfg`)
+
+```
+# Existing LDAP plugin (unchanged, backward compatible)
+ldap_auth_plugin="/path/to/ldap.so"
+
+# Per-user auth plugins (comma-separated paths)
+auth_plugins="/usr/lib/proxysql/auth_static.so"
+```
+
+## Plugin Interface
+
+Plugins implement the `ProxySQL_Auth_Plugin` interface defined in `include/MySQL_AuthPlugin.h`.
+
+### Interface Definition
+
+```cpp
+struct ProxySQL_Auth_Result {
+    bool success;              // true if authentication succeeded
+    char* backend_username;    // optional: map to different backend user (caller frees)
+    char* error_msg;           // optional: error message on failure (caller frees)
+};
+
+class ProxySQL_Auth_Plugin {
+public:
+    virtual ~ProxySQL_Auth_Plugin() {}
+
+    // Initialize the plugin (called once on load)
+    virtual bool init() { return true; }
+
+    // Cleanup the plugin (called once on unload)
+    virtual void deinit() {}
+
+    // Validate user credentials
+    // Returns ProxySQL_Auth_Result with success status and optional backend_username
+    virtual ProxySQL_Auth_Result validate(
+        const char* username,     // Username from client
+        const char* credential,   // Clear-text credential (password, token, etc.)
+        const char* attributes    // JSON attributes from mysql_users.attributes
+    ) = 0;
+
+    // Get plugin name (must match "auth_plugin" value in attributes)
+    virtual const char* name() = 0;
+
+    // Print version info (called on load)
+    virtual void print_version() = 0;
+};
+```
+
+### Required Exports
+
+Each plugin `.so` must export:
+
+```cpp
+extern "C" {
+    // Create plugin instance
+    ProxySQL_Auth_Plugin* proxysql_mysql_auth_plugin_create();
+
+    // Destroy plugin instance
+    void proxysql_mysql_auth_plugin_destroy(ProxySQL_Auth_Plugin* plugin);
+}
+```
+
+## Included Plugins
+
+### Static Auth Plugin (`auth_static.so`)
+
+For testing and simple use cases. Validates credentials against a static password in attributes.
+
+```sql
+INSERT INTO mysql_users (username, password, attributes, frontend, backend)
+VALUES ('testuser', '',
+        '{"auth_plugin": "static", "static_password": "secret123", "backend_username": "dbuser"}',
+        1, 0);
+```
+
+**Source**: `plugins/MySQL_AuthPlugin/static/auth_static.cpp`
+
+### Kubernetes Auth Plugin (`auth_k8s.so`)
+
+Validates Kubernetes ServiceAccount JWT tokens via the K8s TokenReview API.
+
+```sql
+INSERT INTO mysql_users (username, password, attributes, frontend, backend)
+VALUES ('proxysql/myapp', '',
+        '{"auth_plugin": "k8s", "backend_username": "dbuser"}',
+        1, 0);
+```
+
+Username format: `<namespace>/<serviceaccount>` - must match the token's identity.
+
+**Environment Variables** (optional, for overriding in-cluster defaults):
+- `K8S_API_SERVER` - K8s API server URL (default: `https://kubernetes.default.svc`)
+- `K8S_CA_PATH` - Path to CA cert (default: `/var/run/secrets/kubernetes.io/serviceaccount/ca.crt`)
+- `K8S_TOKEN_PATH` - Path to SA token (default: `/var/run/secrets/kubernetes.io/serviceaccount/token`)
+- `K8S_TIMEOUT_MS` - HTTP request timeout in milliseconds (default: `5000`)
+
+**Source**: `plugins/MySQL_AuthPlugin/k8s/`
+
+## Implementation Details
+
+### Plugin Loading
+
+**File**: `src/main.cpp` - `LoadPlugins()`
+
+Plugins are loaded on startup from the comma-separated `auth_plugins` config value:
+
+```cpp
+std::map<std::string, ProxySQL_Auth_Plugin*> GloAuthPlugins;
+
+// For each plugin path:
+void* handle = dlopen(plugin_path.c_str(), RTLD_NOW);
+auto create_func = (proxysql_mysql_auth_plugin_create_t)
+    dlsym(handle, "proxysql_mysql_auth_plugin_create");
+ProxySQL_Auth_Plugin* plugin = create_func();
+plugin->init();
+GloAuthPlugins[plugin->name()] = plugin;
+```
+
+### Authentication Flow
+
+**File**: `lib/MySQL_Protocol.cpp` - `PPHR_1()`
+
+```cpp
+// 1. Lookup user in mysql_users
+account_details = GloMyAuth->lookup(username, USERNAME_FRONTEND, dup_details);
+
+// 2. Check for per-user auth plugin
+ProxySQL_Auth_Plugin* user_auth_plugin = get_user_auth_plugin(account_details.attributes, username);
+
+if (user_auth_plugin) {
+    if (switching_auth_stage == 0) {
+        // 3. Send AUTH_SWITCH to get clear-text credential
+        generate_pkt_auth_switch_request(true, NULL, NULL);
+        return;
+    }
+
+    // 4. After AUTH_SWITCH, validate via plugin
+    ProxySQL_Auth_Result result = user_auth_plugin->validate(username, password, attributes);
+
+    if (result.success) {
+        // 5. Auth successful - use backend_username if provided
+        if (result.backend_username) {
+            // Map to different backend user
+        }
+    }
+}
+```
+
+## Authentication Flow Diagram
+
+```
+┌─────────────────────────────────────────────────────────────────┐
+│ 1. Client connects                                              │
+│    Username: "app_user"                                         │
+│    Password: <initial - may be hashed>                          │
+└─────────────────────────────────────────────────────────────────┘
+                                │
+                                ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ 2. ProxySQL: Lookup user in mysql_users                         │
+│    Found: attributes = '{"auth_plugin": "static", ...}'         │
+└─────────────────────────────────────────────────────────────────┘
+                                │
+                                ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ 3. ProxySQL: Check GloAuthPlugins["static"]                     │
+│    - Not found? → Error: "Auth plugin 'static' not loaded"      │
+│    - Found? → Continue                                          │
+└─────────────────────────────────────────────────────────────────┘
+                                │
+                                ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ 4. ProxySQL: AUTH_SWITCH to mysql_clear_password                │
+│    Sends: 0xFE + "mysql_clear_password" + 0x00                  │
+└─────────────────────────────────────────────────────────────────┘
+                                │
+                                ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ 5. Client: Resends password in clear text                       │
+└─────────────────────────────────────────────────────────────────┘
+                                │
+                                ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ 6. ProxySQL: Call plugin->validate(username, password, attrs)   │
+│    Plugin compares password against static_password in attrs    │
+└─────────────────────────────────────────────────────────────────┘
+                                │
+                    ┌───────────┴───────────┐
+                    ▼                       ▼
+┌───────────────────────────┐ ┌───────────────────────────────────┐
+│ 7a. Validation SUCCESS    │ │ 7b. Validation FAILED             │
+│     - Auth complete       │ │     - Return auth error to client │
+│     - Use backend_username│ │                                   │
+└───────────────────────────┘ └───────────────────────────────────┘
+                    │
+                    ▼
+┌─────────────────────────────────────────────────────────────────┐
+│ 8. ProxySQL: Connect to backend with mapped credentials         │
+│    Frontend "app_user" → Backend "dbuser"                       │
+└─────────────────────────────────────────────────────────────────┘
+```
+
+## Directory Structure
+
+```
+plugins/MySQL_AuthPlugin/
+├── static/
+│   ├── auth_static.cpp      # Static password auth plugin
+│   └── Makefile
+└── k8s/
+    ├── auth_k8s.cpp         # K8s auth plugin entry point
+    ├── k8s_auth_plugin.h    # Plugin class (testable)
+    ├── kubernetes_client.h  # Abstract K8s client interface
+    ├── k8s_http_client.h    # Production HTTP client
+    ├── mock_kubernetes_client.h  # Mock for unit tests
+    └── Makefile
+```
+
+## Testing
+
+### Integration Tests
+
+Static plugin has integration tests requiring ProxySQL and MySQL:
+
+```bash
+# Run integration tests
+./test/tap/tests/run_mysql_authplugin_test.sh
+```
+
+**Source**: `test/tap/tests/test_mysql_authplugin-t.cpp`
+
+### Unit Tests (K8s Plugin)
+
+K8s plugin has unit tests using mock dependency injection:
+
+```bash
+# Build and run unit tests
+make build_tap_tests
+./test/tap/tests/unit-k8s_auth_plugin-t
+```
+
+**Source**: `test/tap/tests/unit-k8s_auth_plugin-t.cpp`
+
+## Backward Compatibility
+
+| Aspect | Behavior |
+|--------|----------|
+| Existing `ldap_auth_plugin` config | Unchanged, still works |
+| Existing `GloMyLdapAuth` global | Unchanged, still works |
+| Users without `auth_plugin` attribute | Standard password auth |
+| LDAP "user not found" flow | Unchanged |
+| Existing mysql_users entries | No migration needed |
+
+## Security Considerations
+
+1. **Clear-text tokens**: Tokens are sent in clear after AUTH_SWITCH. TLS should be required for production.
+2. **Token validation**: Plugins should validate tokens against trusted sources (K8s API, OAuth provider).
+3. **Backend credentials**: Backend users should have strong passwords and limited privileges.
+4. **Plugin trust**: Only load plugins from trusted paths.
+
+## Future Enhancements
+
+1. **Admin commands**: `SHOW AUTH PLUGINS` to list loaded plugins
+2. **Runtime reload**: Reload plugins without restart
+3. **Plugin variables**: Per-plugin configuration via admin interface
+4. **Metrics**: Authentication stats per plugin
+
+## References
+
+- [ProxySQL Users Configuration](https://proxysql.com/documentation/users-configuration/)
+- [ProxySQL Password Management](https://proxysql.com/documentation/password-management/)
+- [MySQL Authentication Plugins](https://dev.mysql.com/doc/refman/8.0/en/pluggable-authentication.html)
+- [Kubernetes ServiceAccount Tokens](https://kubernetes.io/docs/reference/access-authn-authz/authentication/#service-account-tokens)
+- [Kubernetes TokenReview API](https://kubernetes.io/docs/reference/kubernetes-api/authentication-resources/token-review-v1/)
diff --git a/include/MySQL_AuthPlugin.h b/include/MySQL_AuthPlugin.h
new file mode 100644
index 0000000000..7cddb1610a
--- /dev/null
+++ b/include/MySQL_AuthPlugin.h
@@ -0,0 +1,76 @@
+/**
+ * @file MySQL_AuthPlugin.h
+ * @brief Interface for ProxySQL per-user authentication plugins (MySQL only)
+ *
+ * Plugins are configured per-user via the `attributes` JSON field in mysql_users.
+ *
+ * Example:
+ *   INSERT INTO mysql_users (username, attributes, frontend, backend)
+ *   VALUES ('myuser', '{"auth_plugin": "myplugin", "backend_username": "dbuser", ...}', 1, 0);
+ */
+
+#ifndef MYSQL_AUTHPLUGIN_H
+#define MYSQL_AUTHPLUGIN_H
+
+#include <cstddef>
+
+/**
+ * @brief Result of authentication validation
+ */
+struct ProxySQL_Auth_Result {
+    bool success;              // true if authentication succeeded
+    char* backend_username;    // optional: map to different backend user (caller frees, NULL = same user)
+    char* error_msg;           // optional: error message on failure (caller frees, NULL = generic error)
+};
+
+/**
+ * @brief Base class for ProxySQL authentication plugins
+ */
+class ProxySQL_Auth_Plugin {
+public:
+    virtual ~ProxySQL_Auth_Plugin() {}
+
+    /**
+     * @brief Initialize the plugin (called once on load)
+     * @return true on success
+     */
+    virtual bool init() { return true; }
+
+    /**
+     * @brief Cleanup the plugin (called once on unload)
+     */
+    virtual void deinit() {}
+
+    /**
+     * @brief Validate user credentials
+     *
+     * @param username    Username from client
+     * @param credential  Clear-text credential from client (password, token, etc.)
+     * @param attributes  JSON attributes from mysql_users.attributes
+     *
+     * @return ProxySQL_Auth_Result with success status and optional backend_username
+     */
+    virtual ProxySQL_Auth_Result validate(
+        const char* username,
+        const char* credential,
+        const char* attributes
+    ) = 0;
+
+    /**
+     * @brief Get plugin name
+     */
+    virtual const char* name() = 0;
+
+    /**
+     * @brief Print version info (called on load)
+     */
+    virtual void print_version() = 0;
+};
+
+// Plugin factory function type
+extern "C" {
+    typedef ProxySQL_Auth_Plugin* (*proxysql_mysql_auth_plugin_create_t)();
+    typedef void (*proxysql_mysql_auth_plugin_destroy_t)(ProxySQL_Auth_Plugin*);
+}
+
+#endif /* MYSQL_AUTHPLUGIN_H */
diff --git a/include/MySQL_Session.h b/include/MySQL_Session.h
index 45c6231f4d..0e58e5464a 100644
--- a/include/MySQL_Session.h
+++ b/include/MySQL_Session.h
@@ -419,6 +419,7 @@ class MySQL_Session: public Base_Session<MySQL_Session, MySQL_Data_Stream, MySQL
 
 	ProxySQL_Node_Address * proxysql_node_address; // this is used ONLY for Admin, and only if the other party is another proxysql instance part of a cluster
 	bool use_ldap_auth;
+	bool use_auth_plugin; // set when using per-user auth plugin (auth_plugins config)
 	// Fast forward grace close flags: track backend closure during fast forward mode
 	// to allow pending client data to drain before closing the session.
 	bool backend_closed_in_fast_forward;
diff --git a/include/proxysql_glovars.hpp b/include/proxysql_glovars.hpp
index c5b52b2db1..a409c04684 100644
--- a/include/proxysql_glovars.hpp
+++ b/include/proxysql_glovars.hpp
@@ -79,6 +79,7 @@ class ProxySQL_GlobalVariables {
 	char * sqlite3_plugin;
 	char * web_interface_plugin;
 	char * ldap_auth_plugin;
+	char * auth_plugins;
 	SSL_CTX *get_SSL_ctx();
 	SSL *get_SSL_new();
 	void get_SSL_pem_mem(char **key, char **cert);
diff --git a/lib/MySQL_Protocol.cpp b/lib/MySQL_Protocol.cpp
index f83ff07005..4a35e2a19f 100644
--- a/lib/MySQL_Protocol.cpp
+++ b/lib/MySQL_Protocol.cpp
@@ -13,13 +13,16 @@ using json = nlohmann::json;
 #include "MySQL_Authentication.hpp"
 #include "MySQL_LDAP_Authentication.hpp"
 #include "MySQL_Variables.h"
+#include "MySQL_AuthPlugin.h"
 
 #include <sstream>
+#include <map>
 
 //#include <ma_global.h>
 
 extern MySQL_Authentication *GloMyAuth;
 extern MySQL_LDAP_Authentication *GloMyLdapAuth;
+extern std::map<std::string, ProxySQL_Auth_Plugin*> GloAuthPlugins;
 extern MySQL_Threads_Handler *GloMTH;
 
 #ifdef PROXYSQLCLICKHOUSE
@@ -100,6 +103,36 @@ void debug_spiffe_id(const unsigned char *user, const char *attributes, int __li
 }
 #endif
 
+// Helper function to get auth plugin from user attributes
+static ProxySQL_Auth_Plugin* get_user_auth_plugin(const char* attributes, const char* username) {
+	if (!attributes || attributes[0] == '\0') {
+		return nullptr;
+	}
+
+	try {
+		nlohmann::json attrs = nlohmann::json::parse(attributes);
+		auto it = attrs.find("auth_plugin");
+		if (it == attrs.end()) {
+			return nullptr;
+		}
+
+		std::string plugin_name = it->get<std::string>();
+		auto plugin_it = GloAuthPlugins.find(plugin_name);
+		if (plugin_it == GloAuthPlugins.end()) {
+			proxy_error("Auth plugin '%s' specified for user '%s' but not loaded\n",
+				plugin_name.c_str(), username ? username : "unknown");
+			return nullptr;
+		}
+
+		proxy_debug(PROXY_DEBUG_MYSQL_AUTH, 5, "Using auth plugin '%s' for user '%s'\n",
+			plugin_name.c_str(), username ? username : "unknown");
+		return plugin_it->second;
+	} catch (nlohmann::json::exception& e) {
+		proxy_debug(PROXY_DEBUG_MYSQL_AUTH, 5, "Failed to parse attributes JSON for user '%s': %s\n",
+			username ? username : "unknown", e.what());
+		return nullptr;
+	}
+}
 
 void MySQL_Protocol::init(MySQL_Data_Stream **__myds, MySQL_Connection_userinfo *__userinfo, MySQL_Session *__sess) {
 	myds=__myds;
@@ -2347,6 +2380,7 @@ bool MySQL_Protocol::process_pkt_handshake_response(unsigned char *pkt, unsigned
 	MyProt_tmp_auth_vars vars1;
 	account_details_t account_details {};
 	dup_account_details_t dup_details { true, true, true };
+	ProxySQL_Auth_Plugin* user_auth_plugin = nullptr;
 
 	vars1._ptr = pkt;
 	mysql_hdr hdr;
@@ -2467,6 +2501,113 @@ bool MySQL_Protocol::process_pkt_handshake_response(unsigned char *pkt, unsigned
 		account_details = GloMyAuth->lookup((char*)vars1.user, USERNAME_FRONTEND, dup_details);
 	}
 
+	// Check for per-user auth plugin
+	user_auth_plugin = get_user_auth_plugin(account_details.attributes, (const char*)vars1.user);
+	if (user_auth_plugin) {
+		if ((*myds)->switching_auth_stage == 0) {
+			// Need to switch to clear password to get the token/password
+			(*myds)->switching_auth_type = AUTH_MYSQL_CLEAR_PASSWORD;
+			(*myds)->switching_auth_stage = 1;
+			(*myds)->auth_in_progress = 1;
+			generate_pkt_auth_switch_request(true, NULL, NULL);
+			(*myds)->myconn->userinfo->set((char *)vars1.user, NULL, vars1.db, NULL);
+			ret = false;
+			proxy_debug(PROXY_DEBUG_MYSQL_AUTH, 5,
+				"Session=%p , DS=%p , user='%s' . AUTH_SWITCH for per-user auth plugin\n",
+				(*myds), (*myds)->sess, vars1.user);
+			goto __exit_process_pkt_handshake_response;
+		}
+
+		// After AUTH_SWITCH, we have clear-text credential in vars1.pass
+		// Validate via the user's auth plugin
+		ProxySQL_Auth_Result auth_result = user_auth_plugin->validate(
+			(const char *)vars1.user,
+			(const char *)vars1.pass,
+			account_details.attributes
+		);
+
+		if (auth_result.success) {
+			// Auth successful
+			ret = true;
+			(*myds)->sess->default_hostgroup = account_details.default_hostgroup;
+			// Must strdup these as account_details will be freed
+			if ((*myds)->sess->default_schema) {
+				free((*myds)->sess->default_schema);
+			}
+			(*myds)->sess->default_schema = account_details.default_schema ? strdup(account_details.default_schema) : nullptr;
+			if ((*myds)->sess->user_attributes) {
+				free((*myds)->sess->user_attributes);
+			}
+			(*myds)->sess->user_attributes = account_details.attributes ? strdup(account_details.attributes) : nullptr;
+			(*myds)->sess->schema_locked = account_details.schema_locked;
+			(*myds)->sess->transaction_persistent = account_details.transaction_persistent;
+			(*myds)->sess->session_fast_forward = account_details.fast_forward ? SESSION_FORWARD_TYPE_PERMANENT : SESSION_FORWARD_TYPE_NONE;
+			(*myds)->sess->user_max_connections = account_details.max_connections;
+
+			// Use auth plugin path for connection tracking
+			// This ensures the frontend username is tracked separately from backend username
+			(*myds)->sess->use_auth_plugin = true;
+
+			// Handle backend username mapping if provided
+			if (auth_result.backend_username) {
+				account_details_t backend_acct = GloMyAuth->lookup(auth_result.backend_username, USERNAME_BACKEND, { true, true, true });
+				if (backend_acct.password) {
+					// Preserve frontend username for connection tracking
+					userinfo->fe_username = strdup((const char *)vars1.user);
+					// Set backend username and password for backend connection
+					userinfo->set(auth_result.backend_username, backend_acct.password, NULL, NULL);
+					// Also update vars1 so the code at __exit_do_auth doesn't overwrite our settings
+					if (vars1.password) free(vars1.password);
+					vars1.password = strdup(backend_acct.password);
+					// Set SHA1 password (hex format) for backend authentication
+					if (userinfo->sha1_pass) free(userinfo->sha1_pass);
+					userinfo->sha1_pass = NULL;
+					if (backend_acct.sha1_pass) {
+						// sha1_pass is already in hex format (like "*ABC123...")
+						userinfo->sha1_pass = strdup((char*)backend_acct.sha1_pass);
+					} else if (backend_acct.password[0] != '*') {
+						// Password is clear-text, compute SHA1 and convert to hex
+						char sha1_binary[SHA_DIGEST_LENGTH];
+						size_t pass_len = strlen(backend_acct.password); // safe: password checked non-null at line 2554
+						SHA1((const unsigned char*)backend_acct.password,
+							pass_len,
+							(unsigned char*)sha1_binary);
+						userinfo->sha1_pass = sha1_pass_hex(sha1_binary);
+						// Cache the computed SHA1 for future use (in binary format for set_SHA1)
+						GloMyAuth->set_SHA1(auth_result.backend_username, USERNAME_BACKEND, sha1_binary);
+					}
+					proxy_debug(PROXY_DEBUG_MYSQL_AUTH, 5,
+						"Session=%p , DS=%p , frontend_user='%s' mapped to backend_user='%s'\n",
+						(*myds), (*myds)->sess, vars1.user, auth_result.backend_username);
+				} else {
+					proxy_error("Backend user '%s' not found for frontend user '%s'\n",
+						auth_result.backend_username, vars1.user);
+					ret = false;
+				}
+				free_account_details(backend_acct);
+			} else {
+				// No backend mapping, use frontend username for connection tracking
+				userinfo->fe_username = strdup((const char *)vars1.user);
+			}
+		} else {
+			// Auth failed
+			ret = false;
+			if (auth_result.error_msg) {
+				proxy_debug(PROXY_DEBUG_MYSQL_AUTH, 5,
+					"Session=%p , DS=%p , user='%s' . Auth plugin error: %s\n",
+					(*myds), (*myds)->sess, vars1.user, auth_result.error_msg);
+			} else {
+				proxy_debug(PROXY_DEBUG_MYSQL_AUTH, 5,
+					"Session=%p , DS=%p , user='%s' . Per-user auth plugin validation failed\n",
+					(*myds), (*myds)->sess, vars1.user);
+			}
+		}
+		// Cleanup auth result
+		if (auth_result.backend_username) free(auth_result.backend_username);
+		if (auth_result.error_msg) free(auth_result.error_msg);
+		goto __exit_do_auth;
+	}
+
 	vars1.password = get_password(account_details, PASSWORD_TYPE::PRIMARY);
 	vars1.passtype = PASSWORD_TYPE::PRIMARY;
 	// the async state machine needs to change; we are creating overhead in auth for old-passwords
diff --git a/lib/MySQL_Session.cpp b/lib/MySQL_Session.cpp
index 6e4bc4cd5c..f12fd0d096 100644
--- a/lib/MySQL_Session.cpp
+++ b/lib/MySQL_Session.cpp
@@ -689,6 +689,7 @@ MySQL_Session::MySQL_Session() {
 	last_HG_affected_rows = -1; // #1421 : advanced support for LAST_INSERT_ID()
 	proxysql_node_address = NULL;
 	use_ldap_auth = false;
+	use_auth_plugin = false;
 	this->wait_timeout = mysql_thread___wait_timeout;
 	backend_closed_in_fast_forward = false;
 	fast_forward_grace_start_time = 0;
@@ -765,12 +766,18 @@ MySQL_Session::~MySQL_Session() {
 					break;
 #endif /* PROXYSQLCLICKHOUSE */
 				default:
-					if (use_ldap_auth == false) {
+					if (use_auth_plugin) {
+						// Per-user auth plugin path: use GloMyAuth with fe_username
+						GloMyAuth->decrease_frontend_user_connections(
+							client_myds->myconn->userinfo->fe_username,
+							PASSWORD_TYPE::PRIMARY
+						);
+					} else if (use_ldap_auth == false) {
 						GloMyAuth->decrease_frontend_user_connections(
 							client_myds->myconn->userinfo->username,
 							client_myds->myconn->userinfo->passtype
 						);
-					} else {
+					} else if (GloMyLdapAuth) {
 						GloMyLdapAuth->decrease_frontend_user_connections(client_myds->myconn->userinfo->fe_username);
 					}
 					break;
@@ -5644,13 +5651,21 @@ void MySQL_Session::handler___status_CONNECTING_CLIENT___STATE_SERVER_HANDSHAKE(
 //#endif // TEST_AURORA || TEST_GALERA || TEST_GROUPREP
 					case PROXYSQL_SESSION_MYSQL:
 						proxy_debug(PROXY_DEBUG_MYSQL_CONNECTION,8,"Session=%p , DS=%p , session_type=PROXYSQL_SESSION_MYSQL\n", this, client_myds);
-						if (use_ldap_auth == false) {
+						if (use_auth_plugin) {
+							// Per-user auth plugin path: use GloMyAuth with fe_username
+							// fe_username contains the frontend user that was authenticated via plugin
+							free_users = GloMyAuth->increase_frontend_user_connections(
+								client_myds->myconn->userinfo->fe_username,
+								PASSWORD_TYPE::PRIMARY,
+								&used_users
+							);
+						} else if (use_ldap_auth == false) {
 							free_users = GloMyAuth->increase_frontend_user_connections(
 								client_myds->myconn->userinfo->username,
 								client_myds->myconn->userinfo->passtype,
 								&used_users
 							);
-						} else {
+						} else if (GloMyLdapAuth) {
 							free_users = GloMyLdapAuth->increase_frontend_user_connections(client_myds->myconn->userinfo->fe_username, &used_users);
 						}
 						break;
@@ -7272,12 +7287,18 @@ void MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C
 		reset();
 		init();
 		if (client_authenticated) {
-			if (use_ldap_auth == false) {
+			if (use_auth_plugin) {
+				// Per-user auth plugin path: use GloMyAuth with fe_username
+				GloMyAuth->decrease_frontend_user_connections(
+					client_myds->myconn->userinfo->fe_username,
+					PASSWORD_TYPE::PRIMARY
+				);
+			} else if (use_ldap_auth == false) {
 				GloMyAuth->decrease_frontend_user_connections(
 					client_myds->myconn->userinfo->username,
 					client_myds->myconn->userinfo->passtype
 				);
-			} else {
+			} else if (GloMyLdapAuth) {
 				GloMyLdapAuth->decrease_frontend_user_connections(client_myds->myconn->userinfo->fe_username);
 			}
 		}
diff --git a/lib/ProxySQL_GloVars.cpp b/lib/ProxySQL_GloVars.cpp
index f0da1a812a..8027817267 100644
--- a/lib/ProxySQL_GloVars.cpp
+++ b/lib/ProxySQL_GloVars.cpp
@@ -97,6 +97,10 @@ ProxySQL_GlobalVariables::~ProxySQL_GlobalVariables() {
 		free(ldap_auth_plugin);
 		ldap_auth_plugin = NULL;
 	}
+	if (auth_plugins) {
+		free(auth_plugins);
+		auth_plugins = NULL;
+	}
 	/**
 	 * @brief set in_shutdown flag just the member 'checksums_values'.
 	 * @details This is performed to prevent the free() inside the 'ProxySQL_Checksum_Value' destructor for
@@ -216,6 +220,7 @@ ProxySQL_GlobalVariables::ProxySQL_GlobalVariables() :
 	checksums_values.global_checksum = 0;
 	execute_on_exit_failure = NULL;
 	ldap_auth_plugin = NULL;
+	auth_plugins = NULL;
 	web_interface_plugin = NULL;
 	sqlite3_plugin = NULL;
 #ifdef DEBUG
diff --git a/plugins/MySQL_AuthPlugin/k8s/.gitignore b/plugins/MySQL_AuthPlugin/k8s/.gitignore
new file mode 100644
index 0000000000..cc7836862d
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/.gitignore
@@ -0,0 +1,2 @@
+*.so
+*.o
diff --git a/plugins/MySQL_AuthPlugin/k8s/Makefile b/plugins/MySQL_AuthPlugin/k8s/Makefile
new file mode 100644
index 0000000000..c0c6620719
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/Makefile
@@ -0,0 +1,30 @@
+# ProxySQL K8s Auth Plugin Makefile
+
+PROXYSQL_ROOT := ../../..
+CXX := g++
+CXXFLAGS := -std=c++17 -fPIC -O2 -Wall
+INCLUDES := -I$(PROXYSQL_ROOT)/include -I$(PROXYSQL_ROOT)/deps/json -I$(PROXYSQL_ROOT)/deps/curl/curl/include
+CURL_LIB := $(PROXYSQL_ROOT)/deps/curl/curl/lib/.libs/libcurl.a
+# curl dependencies (for static linking)
+LDFLAGS := -lssl -lcrypto -lz -lpthread
+
+TARGET := auth_k8s.so
+SOURCES := auth_k8s.cpp
+
+.PHONY: all clean install
+
+all: $(TARGET)
+
+$(TARGET): $(SOURCES)
+	$(CXX) $(CXXFLAGS) $(INCLUDES) -shared -o $@ $< $(CURL_LIB) $(LDFLAGS)
+
+clean:
+	rm -f $(TARGET)
+
+install: $(TARGET)
+	@echo "Install location: specify with 'make install DESTDIR=/path/to/plugins'"
+	@if [ -n "$(DESTDIR)" ]; then \
+		mkdir -p $(DESTDIR); \
+		cp $(TARGET) $(DESTDIR)/; \
+		echo "Installed $(TARGET) to $(DESTDIR)"; \
+	fi
diff --git a/plugins/MySQL_AuthPlugin/k8s/README.md b/plugins/MySQL_AuthPlugin/k8s/README.md
new file mode 100644
index 0000000000..4ce23e1948
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/README.md
@@ -0,0 +1,85 @@
+# K8s Auth Plugin
+
+Kubernetes ServiceAccount authentication plugin for ProxySQL.
+
+## Overview
+
+Validates Kubernetes ServiceAccount JWT tokens via the K8s TokenReview API. Clients send their JWT token as the MySQL password.
+
+## Configuration
+
+### Environment Variables
+
+Override in-cluster defaults:
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `K8S_API_SERVER` | `https://kubernetes.default.svc` | K8s API server URL |
+| `K8S_CA_PATH` | `/var/run/secrets/kubernetes.io/serviceaccount/ca.crt` | Path to CA cert |
+| `K8S_TOKEN_PATH` | `/var/run/secrets/kubernetes.io/serviceaccount/token` | Path to ProxySQL's SA token |
+
+### ProxySQL Config
+
+```
+mysql_variables=
+{
+    auth_plugins="/path/to/auth_k8s.so"
+}
+```
+
+### User Setup
+
+```sql
+INSERT INTO mysql_users (username, password, attributes, frontend, backend)
+VALUES ('k8s-user', '',
+        '{"auth_plugin": "k8s", "backend_username": "dbuser"}',
+        1, 0);
+LOAD MYSQL USERS TO RUNTIME;
+```
+
+### Attributes
+
+| Attribute | Required | Description |
+|-----------|----------|-------------|
+| `auth_plugin` | Yes | Must be `"k8s"` |
+| `backend_username` | No | Backend MySQL user to map to |
+
+## Kubernetes RBAC
+
+ProxySQL's ServiceAccount needs `tokenreviews` permission:
+
+```yaml
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: tokenreview
+rules:
+- apiGroups: ["authentication.k8s.io"]
+  resources: ["tokenreviews"]
+  verbs: ["create"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: proxysql-tokenreview
+subjects:
+- kind: ServiceAccount
+  name: proxysql
+  namespace: default
+roleRef:
+  kind: ClusterRole
+  name: tokenreview
+  apiGroup: rbac.authorization.k8s.io
+```
+
+## Building
+
+```bash
+cd plugins/MySQL_AuthPlugin/k8s
+make
+```
+
+## Dependencies
+
+- libcurl (for HTTP requests)
+- nlohmann/json (included in ProxySQL deps)
diff --git a/plugins/MySQL_AuthPlugin/k8s/auth_k8s.cpp b/plugins/MySQL_AuthPlugin/k8s/auth_k8s.cpp
new file mode 100644
index 0000000000..b8615beac2
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/auth_k8s.cpp
@@ -0,0 +1,43 @@
+/**
+ * @file auth_k8s.cpp
+ * @brief Kubernetes ServiceAccount authentication plugin for ProxySQL
+ *
+ * Validates Kubernetes ServiceAccount JWT tokens via K8s TokenReview API.
+ * Username must be in "namespace/serviceaccount" format and must match the token identity.
+ *
+ * Configuration:
+ *   Environment variables (optional, for overriding in-cluster defaults):
+ *     K8S_API_SERVER  - K8s API server URL (default: https://kubernetes.default.svc)
+ *     K8S_CA_PATH     - Path to CA cert (default: /var/run/secrets/kubernetes.io/serviceaccount/ca.crt)
+ *     K8S_TOKEN_PATH  - Path to SA token (default: /var/run/secrets/kubernetes.io/serviceaccount/token)
+ *
+ *   User attributes example:
+ *     INSERT INTO mysql_users (username, password, attributes, frontend, backend)
+ *     VALUES ('proxysql/testuser', '',
+ *             '{"auth_plugin": "k8s", "backend_username": "dbuser"}',
+ *             1, 0);
+ *
+ * The client sends their ServiceAccount JWT token as the password.
+ * The username "proxysql/testuser" must match token's namespace and serviceaccount.
+ */
+
+#include "k8s_auth_plugin.h"
+#include "k8s_http_client.h"
+
+// Production plugin uses K8sHttpClient as the KubernetesClient implementation
+class K8s_Auth_Plugin_Production : public K8s_Auth_Plugin {
+public:
+    K8s_Auth_Plugin_Production()
+        : K8s_Auth_Plugin(std::make_unique<K8sHttpClient>()) {}
+};
+
+// Plugin exports
+extern "C" {
+    ProxySQL_Auth_Plugin* proxysql_mysql_auth_plugin_create() {
+        return new K8s_Auth_Plugin_Production();
+    }
+
+    void proxysql_mysql_auth_plugin_destroy(ProxySQL_Auth_Plugin* plugin) {
+        delete plugin;
+    }
+}
diff --git a/plugins/MySQL_AuthPlugin/k8s/k8s_auth_plugin.h b/plugins/MySQL_AuthPlugin/k8s/k8s_auth_plugin.h
new file mode 100644
index 0000000000..379a38feaf
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/k8s_auth_plugin.h
@@ -0,0 +1,132 @@
+/**
+ * @file k8s_auth_plugin.h
+ * @brief Kubernetes ServiceAccount authentication plugin class definition
+ *
+ * Separated into header for testability via dependency injection.
+ */
+
+#ifndef K8S_AUTH_PLUGIN_H
+#define K8S_AUTH_PLUGIN_H
+
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <string>
+#include <memory>
+
+#include "../../../include/MySQL_AuthPlugin.h"
+#include "../../../deps/json/json.hpp"
+#include "kubernetes_client.h"
+
+using json = nlohmann::json;
+
+class K8s_Auth_Plugin : public ProxySQL_Auth_Plugin {
+private:
+    std::unique_ptr<KubernetesClient> k8s_client;
+
+public:
+    // Constructor for dependency injection (testing)
+    explicit K8s_Auth_Plugin(std::unique_ptr<KubernetesClient> client)
+        : k8s_client(std::move(client)) {}
+
+    ~K8s_Auth_Plugin() override = default;
+
+    bool init() override {
+        if (!k8s_client->init()) {
+            return false;
+        }
+        fprintf(stderr, "K8s Auth Plugin initialized\n");
+        return true;
+    }
+
+    void deinit() override {
+        k8s_client->deinit();
+        fprintf(stderr, "K8s Auth Plugin destroyed\n");
+    }
+
+    const char* name() override {
+        return "k8s";
+    }
+
+    void print_version() override {
+        fprintf(stderr, "ProxySQL K8s Auth Plugin v1.0\n");
+    }
+
+    ProxySQL_Auth_Result validate(
+        const char* username,
+        const char* credential,
+        const char* attributes
+    ) override {
+        ProxySQL_Auth_Result result = { false, nullptr, nullptr };
+
+        if (!username || !credential || !attributes) {
+            result.error_msg = strdup("missing required parameters");
+            return result;
+        }
+
+        if (credential[0] == '\0') {
+            result.error_msg = strdup("empty token");
+            return result;
+        }
+
+        try {
+            json attrs = json::parse(attributes);
+
+            // Call TokenReview API
+            TokenReviewResult review = k8s_client->tokenReview(credential);
+
+            if (!review.success) {
+                result.error_msg = strdup(review.error.c_str());
+                return result;
+            }
+
+            if (!review.authenticated) {
+                result.error_msg = strdup("token not authenticated");
+                return result;
+            }
+
+            // Parse token username: system:serviceaccount:<namespace>:<sa>
+            const std::string prefix = "system:serviceaccount:";
+            if (review.username.rfind(prefix, 0) != 0) {
+                result.error_msg = strdup("token is not a ServiceAccount token");
+                return result;
+            }
+
+            std::string sa_identity = review.username.substr(prefix.length());
+
+            // Convert to "namespace/serviceaccount" format for comparison
+            size_t colon_pos = sa_identity.find(':');
+            if (colon_pos == std::string::npos) {
+                result.error_msg = strdup("invalid ServiceAccount identity format");
+                return result;
+            }
+
+            std::string expected_username = sa_identity.substr(0, colon_pos) + "/" + sa_identity.substr(colon_pos + 1);
+
+            // Validate username matches token identity
+            if (expected_username != username) {
+                std::string err = "username mismatch: expected '" + expected_username + "', got '" + username + "'";
+                result.error_msg = strdup(err.c_str());
+                return result;
+            }
+
+            // Authentication successful
+            result.success = true;
+
+            // Check for backend_username in attributes (static mapping)
+            auto backend_it = attrs.find("backend_username");
+            if (backend_it != attrs.end()) {
+                std::string bu = backend_it->get<std::string>();
+                result.backend_username = strdup(bu.c_str());
+            }
+
+            return result;
+
+        } catch (json::exception& e) {
+            result.error_msg = strdup(e.what());
+            return result;
+        }
+    }
+};
+
+#endif /* K8S_AUTH_PLUGIN_H */
diff --git a/plugins/MySQL_AuthPlugin/k8s/k8s_http_client.h b/plugins/MySQL_AuthPlugin/k8s/k8s_http_client.h
new file mode 100644
index 0000000000..2408e49e0f
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/k8s_http_client.h
@@ -0,0 +1,172 @@
+/**
+ * @file k8s_http_client.h
+ * @brief Real Kubernetes client implementation using HTTP/curl
+ */
+
+#ifndef K8S_HTTP_CLIENT_H
+#define K8S_HTTP_CLIENT_H
+
+#include "kubernetes_client.h"
+#include <fstream>
+#include <sstream>
+#include <curl/curl.h>
+#include "../../../deps/json/json.hpp"
+
+using json = nlohmann::json;
+
+// In-cluster defaults
+static const char* DEFAULT_K8S_API_SERVER = "https://kubernetes.default.svc";
+static const char* DEFAULT_K8S_CA_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt";
+static const char* DEFAULT_K8S_TOKEN_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token";
+static const long DEFAULT_K8S_TIMEOUT_MS = 5000;  // 5 seconds
+
+class K8sHttpClient : public KubernetesClient {
+private:
+    bool curl_initialized = false;
+    std::string api_server;
+    std::string ca_path;
+    std::string token_path;
+    std::string sa_token;
+    long timeout_ms = DEFAULT_K8S_TIMEOUT_MS;
+
+    static size_t write_callback(void* contents, size_t size, size_t nmemb, void* userp) {
+        size_t realsize = size * nmemb;
+        std::string* str = static_cast<std::string*>(userp);
+        str->append(static_cast<char*>(contents), realsize);
+        return realsize;
+    }
+
+    bool read_file(const std::string& path, std::string& contents) {
+        std::ifstream file(path);
+        if (!file.is_open()) return false;
+        std::stringstream buffer;
+        buffer << file.rdbuf();
+        contents = buffer.str();
+        return true;
+    }
+
+public:
+    K8sHttpClient() = default;
+    ~K8sHttpClient() override { deinit(); }
+
+    bool init() override {
+        CURLcode res = curl_global_init(CURL_GLOBAL_DEFAULT);
+        if (res != CURLE_OK) {
+            fprintf(stderr, "K8sHttpClient: curl_global_init failed: %s\n", curl_easy_strerror(res));
+            return false;
+        }
+        curl_initialized = true;
+
+        const char* env_val;
+        env_val = getenv("K8S_API_SERVER");
+        api_server = env_val ? env_val : DEFAULT_K8S_API_SERVER;
+
+        env_val = getenv("K8S_CA_PATH");
+        ca_path = env_val ? env_val : DEFAULT_K8S_CA_PATH;
+
+        env_val = getenv("K8S_TOKEN_PATH");
+        token_path = env_val ? env_val : DEFAULT_K8S_TOKEN_PATH;
+
+        env_val = getenv("K8S_TIMEOUT_MS");
+        if (env_val) {
+            timeout_ms = std::atol(env_val);
+            if (timeout_ms <= 0) timeout_ms = DEFAULT_K8S_TIMEOUT_MS;
+        }
+
+        if (!read_file(token_path, sa_token)) {
+            fprintf(stderr, "K8sHttpClient: warning: could not read SA token from %s\n", token_path.c_str());
+        }
+
+        fprintf(stderr, "K8sHttpClient initialized (api_server=%s, timeout_ms=%ld)\n", api_server.c_str(), timeout_ms);
+        return true;
+    }
+
+    void deinit() override {
+        if (curl_initialized) {
+            curl_global_cleanup();
+            curl_initialized = false;
+        }
+    }
+
+    TokenReviewResult tokenReview(const std::string& token) override {
+        TokenReviewResult result = {false, false, "", ""};
+
+        // Refresh SA token (may have been rotated)
+        if (!read_file(token_path, sa_token)) {
+            result.error = "failed to read SA token for API auth";
+            return result;
+        }
+
+        CURL* curl = curl_easy_init();
+        if (!curl) {
+            result.error = "failed to initialize curl";
+            return result;
+        }
+
+        // Build TokenReview request
+        json request_body;
+        request_body["apiVersion"] = "authentication.k8s.io/v1";
+        request_body["kind"] = "TokenReview";
+        request_body["spec"]["token"] = token;
+        std::string body = request_body.dump();
+
+        std::string url = api_server + "/apis/authentication.k8s.io/v1/tokenreviews";
+        std::string response;
+
+        struct curl_slist* headers = nullptr;
+        headers = curl_slist_append(headers, "Content-Type: application/json");
+        std::string auth_header = "Authorization: Bearer " + sa_token;
+        headers = curl_slist_append(headers, auth_header.c_str());
+
+        curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
+        curl_easy_setopt(curl, CURLOPT_POST, 1L);
+        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.c_str());
+        curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
+        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_callback);
+        curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response);
+        curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, timeout_ms);
+        curl_easy_setopt(curl, CURLOPT_CAINFO, ca_path.c_str());
+
+        CURLcode res = curl_easy_perform(curl);
+        long http_code = 0;
+        curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code);
+
+        curl_slist_free_all(headers);
+        curl_easy_cleanup(curl);
+
+        if (res != CURLE_OK) {
+            result.error = std::string("HTTP request failed: ") + curl_easy_strerror(res);
+            return result;
+        }
+
+        if (http_code != 200 && http_code != 201) {
+            result.error = "TokenReview API returned HTTP " + std::to_string(http_code);
+            return result;
+        }
+
+        // Parse response
+        try {
+            json resp = json::parse(response);
+
+            if (!resp.contains("status") || !resp["status"].contains("authenticated")) {
+                result.error = "invalid TokenReview response";
+                return result;
+            }
+
+            result.success = true;
+            result.authenticated = resp["status"]["authenticated"].get<bool>();
+
+            if (result.authenticated && resp["status"].contains("user") &&
+                resp["status"]["user"].contains("username")) {
+                result.username = resp["status"]["user"]["username"].get<std::string>();
+            }
+        } catch (json::exception& e) {
+            result.error = e.what();
+            return result;
+        }
+
+        return result;
+    }
+};
+
+#endif /* K8S_HTTP_CLIENT_H */
diff --git a/plugins/MySQL_AuthPlugin/k8s/kubernetes_client.h b/plugins/MySQL_AuthPlugin/k8s/kubernetes_client.h
new file mode 100644
index 0000000000..9abdf82180
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/kubernetes_client.h
@@ -0,0 +1,51 @@
+/**
+ * @file kubernetes_client.h
+ * @brief Abstract interface for Kubernetes API operations
+ *
+ * Provides an abstraction layer for K8s API calls, enabling:
+ * - Real implementation using HTTP/curl for production
+ * - Mock implementation for unit testing
+ */
+
+#ifndef KUBERNETES_CLIENT_H
+#define KUBERNETES_CLIENT_H
+
+#include <string>
+
+/**
+ * @brief Result of a TokenReview API call
+ */
+struct TokenReviewResult {
+    bool success;           // API call succeeded
+    bool authenticated;     // Token is valid
+    std::string username;   // e.g., "system:serviceaccount:namespace:sa"
+    std::string error;      // Error message if !success
+};
+
+/**
+ * @brief Abstract interface for Kubernetes API operations
+ */
+class KubernetesClient {
+public:
+    virtual ~KubernetesClient() = default;
+
+    /**
+     * @brief Initialize the client
+     * @return true on success
+     */
+    virtual bool init() = 0;
+
+    /**
+     * @brief Cleanup resources
+     */
+    virtual void deinit() = 0;
+
+    /**
+     * @brief Validate a ServiceAccount token via TokenReview API
+     * @param token The JWT token to validate
+     * @return TokenReviewResult with authentication status and user info
+     */
+    virtual TokenReviewResult tokenReview(const std::string& token) = 0;
+};
+
+#endif /* KUBERNETES_CLIENT_H */
diff --git a/plugins/MySQL_AuthPlugin/k8s/mock_kubernetes_client.h b/plugins/MySQL_AuthPlugin/k8s/mock_kubernetes_client.h
new file mode 100644
index 0000000000..cec7abb788
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/k8s/mock_kubernetes_client.h
@@ -0,0 +1,114 @@
+/**
+ * @file mock_kubernetes_client.h
+ * @brief Mock Kubernetes client for unit testing
+ */
+
+#ifndef MOCK_KUBERNETES_CLIENT_H
+#define MOCK_KUBERNETES_CLIENT_H
+
+#include "kubernetes_client.h"
+#include <map>
+#include <thread>
+#include <chrono>
+
+class MockKubernetesClient : public KubernetesClient {
+private:
+    // Map token -> TokenReviewResult
+    std::map<std::string, TokenReviewResult> token_responses;
+    // Map token -> delay in milliseconds
+    std::map<std::string, int> token_delays;
+    TokenReviewResult default_response;
+    int default_delay_ms = 0;
+
+public:
+    MockKubernetesClient() {
+        default_response = {false, false, "", "token not found in mock"};
+    }
+
+    ~MockKubernetesClient() override = default;
+
+    bool init() override {
+        return true;
+    }
+
+    void deinit() override {
+    }
+
+    /**
+     * @brief Add a canned response for a specific token
+     */
+    void addTokenResponse(const std::string& token, const TokenReviewResult& response) {
+        token_responses[token] = response;
+    }
+
+    /**
+     * @brief Add a valid token with specified identity
+     */
+    void addValidToken(const std::string& token, const std::string& ns, const std::string& sa) {
+        TokenReviewResult result;
+        result.success = true;
+        result.authenticated = true;
+        result.username = "system:serviceaccount:" + ns + ":" + sa;
+        token_responses[token] = result;
+    }
+
+    /**
+     * @brief Add an invalid/expired token
+     */
+    void addInvalidToken(const std::string& token) {
+        TokenReviewResult result;
+        result.success = true;
+        result.authenticated = false;
+        token_responses[token] = result;
+    }
+
+    /**
+     * @brief Simulate API error for a token
+     */
+    void addErrorToken(const std::string& token, const std::string& error) {
+        TokenReviewResult result;
+        result.success = false;
+        result.error = error;
+        token_responses[token] = result;
+    }
+
+    /**
+     * @brief Add a slow token that delays response by specified milliseconds
+     */
+    void addSlowToken(const std::string& token, int delay_ms, const std::string& ns, const std::string& sa) {
+        addValidToken(token, ns, sa);
+        token_delays[token] = delay_ms;
+    }
+
+    /**
+     * @brief Set default delay for all responses
+     */
+    void setDefaultDelay(int delay_ms) {
+        default_delay_ms = delay_ms;
+    }
+
+    TokenReviewResult tokenReview(const std::string& token) override {
+        // Check for per-token delay
+        auto delay_it = token_delays.find(token);
+        int delay = (delay_it != token_delays.end()) ? delay_it->second : default_delay_ms;
+
+        if (delay > 0) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(delay));
+        }
+
+        auto it = token_responses.find(token);
+        if (it != token_responses.end()) {
+            return it->second;
+        }
+        return default_response;
+    }
+
+    /**
+     * @brief Clear all canned responses
+     */
+    void clear() {
+        token_responses.clear();
+    }
+};
+
+#endif /* MOCK_KUBERNETES_CLIENT_H */
diff --git a/plugins/MySQL_AuthPlugin/static/.gitignore b/plugins/MySQL_AuthPlugin/static/.gitignore
new file mode 100644
index 0000000000..cc7836862d
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/static/.gitignore
@@ -0,0 +1,2 @@
+*.so
+*.o
diff --git a/plugins/MySQL_AuthPlugin/static/Makefile b/plugins/MySQL_AuthPlugin/static/Makefile
new file mode 100644
index 0000000000..c79c2b1f3f
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/static/Makefile
@@ -0,0 +1,27 @@
+# ProxySQL Static Auth Plugin Makefile
+
+PROXYSQL_ROOT := ../../..
+CXX := g++
+CXXFLAGS := -std=c++17 -fPIC -O2 -Wall
+INCLUDES := -I$(PROXYSQL_ROOT)/include -I$(PROXYSQL_ROOT)/deps/json
+
+TARGET := auth_static.so
+SOURCES := auth_static.cpp
+
+.PHONY: all clean install
+
+all: $(TARGET)
+
+$(TARGET): $(SOURCES)
+	$(CXX) $(CXXFLAGS) $(INCLUDES) -shared -o $@ $<
+
+clean:
+	rm -f $(TARGET)
+
+install: $(TARGET)
+	@echo "Install location: specify with 'make install DESTDIR=/path/to/plugins'"
+	@if [ -n "$(DESTDIR)" ]; then \
+		mkdir -p $(DESTDIR); \
+		cp $(TARGET) $(DESTDIR)/; \
+		echo "Installed $(TARGET) to $(DESTDIR)"; \
+	fi
diff --git a/plugins/MySQL_AuthPlugin/static/README.md b/plugins/MySQL_AuthPlugin/static/README.md
new file mode 100644
index 0000000000..b1589012af
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/static/README.md
@@ -0,0 +1,96 @@
+# ProxySQL Static Auth Plugin
+
+A simple authentication plugin that validates credentials against a static password stored in the user's `attributes` JSON field.
+
+## Use Cases
+
+### 1. ProxySQL-Only Users
+
+Create frontend users that don't exist in upstream MySQL databases. Multiple frontend users can map to a single backend database user:
+
+```
+app-team-a (frontend) ──┐
+                        ├──► app_backend (backend) ──► MySQL
+app-team-b (frontend) ──┘
+```
+
+This enables:
+- **Credential isolation**: Each team/service gets unique credentials without creating MySQL users
+- **Credential rotation**: Rotate frontend passwords without touching the database
+- **Access auditing**: Track which frontend user connected via ProxySQL logs
+
+### 2. Testing and Development
+
+Test the per-user auth plugin system without external dependencies (no K8s, no LDAP, etc.).
+
+## Build
+
+```bash
+cd plugins/MySQL_AuthPlugin/static
+make
+```
+
+This produces `auth_static.so`.
+
+## Configuration
+
+### 1. Configure ProxySQL to load the plugin
+
+In `proxysql.cfg`:
+
+```
+auth_plugins="/path/to/auth_static.so"
+```
+
+### 2. Create a user with static auth
+
+```sql
+INSERT INTO mysql_users (username, password, attributes, default_hostgroup, frontend, backend)
+VALUES (
+    'testuser',
+    '',
+    '{"auth_plugin": "static", "static_password": "secret123"}',
+    1,
+    1,
+    0
+);
+
+LOAD MYSQL USERS TO RUNTIME;
+```
+
+### 3. Connect
+
+```bash
+mysql -h 127.0.0.1 -P 6033 -u testuser -psecret123
+```
+
+## Attributes
+
+| Attribute | Required | Description |
+|-----------|----------|-------------|
+| `auth_plugin` | Yes | Must be `"static"` |
+| `static_password` | Yes | The password to validate against |
+| `backend_username` | No | Map to a different backend user |
+
+## Backend User Mapping
+
+You can map the frontend user to a different backend user:
+
+```sql
+-- Frontend user authenticates with static password
+INSERT INTO mysql_users (username, password, attributes, default_hostgroup, frontend, backend)
+VALUES (
+    'app_frontend',
+    '',
+    '{"auth_plugin": "static", "static_password": "token123", "backend_username": "app_backend"}',
+    1,
+    1,
+    0
+);
+
+-- Backend user for actual database connections
+INSERT INTO mysql_users (username, password, default_hostgroup, frontend, backend)
+VALUES ('app_backend', 'real_db_password', 1, 0, 1);
+
+LOAD MYSQL USERS TO RUNTIME;
+```
diff --git a/plugins/MySQL_AuthPlugin/static/auth_static.cpp b/plugins/MySQL_AuthPlugin/static/auth_static.cpp
new file mode 100644
index 0000000000..c7f27e59a4
--- /dev/null
+++ b/plugins/MySQL_AuthPlugin/static/auth_static.cpp
@@ -0,0 +1,104 @@
+/**
+ * @file auth_static.cpp
+ * @brief Static authentication plugin for ProxySQL
+ *
+ * Validates credentials against a static password in the user's attributes JSON.
+ *
+ * Configuration example:
+ *   INSERT INTO mysql_users (username, password, attributes, frontend, backend)
+ *   VALUES ('testuser', '',
+ *           '{"auth_plugin": "static", "static_password": "secret123", "backend_username": "dbuser"}',
+ *           1, 0);
+ */
+
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <string>
+
+#include "../../../include/MySQL_AuthPlugin.h"
+#include "../../../deps/json/json.hpp"
+
+using json = nlohmann::json;
+
+class Static_Auth_Plugin : public ProxySQL_Auth_Plugin {
+public:
+    Static_Auth_Plugin() {}
+    ~Static_Auth_Plugin() override {}
+
+    bool init() override {
+        fprintf(stderr, "Static Auth Plugin initialized\n");
+        return true;
+    }
+
+    void deinit() override {
+        fprintf(stderr, "Static Auth Plugin destroyed\n");
+    }
+
+    const char* name() override {
+        return "static";
+    }
+
+    void print_version() override {
+        fprintf(stderr, "ProxySQL Static Auth Plugin v1.0\n");
+    }
+
+    ProxySQL_Auth_Result validate(
+        const char* username,
+        const char* credential,
+        const char* attributes
+    ) override {
+        ProxySQL_Auth_Result result = { false, nullptr, nullptr };
+
+        if (!username || !credential || !attributes) {
+            result.error_msg = strdup("missing required parameters");
+            return result;
+        }
+
+        try {
+            json attrs = json::parse(attributes);
+
+            // Get static_password from attributes
+            auto it = attrs.find("static_password");
+            if (it == attrs.end()) {
+                result.error_msg = strdup("'static_password' not found in attributes");
+                return result;
+            }
+
+            std::string expected_password = it->get<std::string>();
+
+            // Compare passwords
+            if (expected_password != credential) {
+                result.error_msg = strdup("password mismatch");
+                return result;
+            }
+
+            // Authentication successful
+            result.success = true;
+
+            // Check for backend_username mapping
+            auto backend_it = attrs.find("backend_username");
+            if (backend_it != attrs.end()) {
+                std::string bu = backend_it->get<std::string>();
+                result.backend_username = strdup(bu.c_str());
+            }
+
+            return result;
+
+        } catch (json::exception& e) {
+            result.error_msg = strdup(e.what());
+            return result;
+        }
+    }
+};
+
+// Plugin exports
+extern "C" {
+    ProxySQL_Auth_Plugin* proxysql_mysql_auth_plugin_create() {
+        return new Static_Auth_Plugin();
+    }
+
+    void proxysql_mysql_auth_plugin_destroy(ProxySQL_Auth_Plugin* plugin) {
+        delete plugin;
+    }
+}
diff --git a/src/main.cpp b/src/main.cpp
index aa78d0f799..4b31b624be 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -6,6 +6,8 @@ using json = nlohmann::json;
 
 #include <iostream>
 #include <thread>
+#include <map>
+#include <sstream>
 #include "btree_map.h"
 #include "proxysql.h"
 
@@ -32,6 +34,7 @@ using json = nlohmann::json;
 #include "MySQL_Authentication.hpp"
 #include "PgSQL_Authentication.h"
 #include "MySQL_LDAP_Authentication.hpp"
+#include "MySQL_AuthPlugin.h"
 #include "MySQL_Query_Cache.h"
 #include "PgSQL_Query_Cache.h"
 #include "proxysql_restapi.h"
@@ -469,6 +472,7 @@ PgSQL_Query_Cache* GloPgQC;
 MySQL_Authentication *GloMyAuth;
 PgSQL_Authentication* GloPgAuth;
 MySQL_LDAP_Authentication *GloMyLdapAuth;
+std::map<std::string, ProxySQL_Auth_Plugin*> GloAuthPlugins;
 #ifdef PROXYSQLCLICKHOUSE
 ClickHouse_Authentication *GloClickHouseAuth;
 #endif /* PROXYSQLCLICKHOUSE */
@@ -786,6 +790,14 @@ void ProxySQL_Main_process_global_variables(int argc, const char **argv) {
 				GloVars.ldap_auth_plugin=strdup(ldap_auth_plugin.c_str());
 			}
 		}
+		if (root.exists("auth_plugins")==true) {
+			string auth_plugins;
+			bool rc;
+			rc=root.lookupValue("auth_plugins", auth_plugins);
+			if (rc==true) {
+				GloVars.auth_plugins=strdup(auth_plugins.c_str());
+			}
+		}
 		const map<string, char**> varnames_globals_map {
 			{ "mysql-ssl_p2s_ca", &GloVars.global.gr_bootstrap_ssl_ca },
 			{ "mysql-ssl_p2s_capath", &GloVars.global.gr_bootstrap_ssl_capath },
@@ -1405,6 +1417,80 @@ static void LoadPlugins() {
 			//}
 		}
 	}
+	// Load additional auth plugins
+	if (GloVars.auth_plugins) {
+		std::string plugins_str(GloVars.auth_plugins);
+		std::stringstream ss(plugins_str);
+		std::string plugin_path;
+
+		while (std::getline(ss, plugin_path, ',')) {
+			// Trim whitespace
+			size_t start = plugin_path.find_first_not_of(" \t");
+			size_t end = plugin_path.find_last_not_of(" \t");
+			if (start == std::string::npos) continue;
+			plugin_path = plugin_path.substr(start, end - start + 1);
+
+			if (plugin_path.empty()) continue;
+
+			dlerror(); // Clear errors
+			void* handle = dlopen(plugin_path.c_str(), RTLD_NOW);
+			if (!handle) {
+				proxy_error("Cannot load auth plugin '%s': %s\n",
+					plugin_path.c_str(), dlerror());
+				continue;
+			}
+
+			// Get factory function
+			dlerror();
+			proxysql_mysql_auth_plugin_create_t create_func =
+				(proxysql_mysql_auth_plugin_create_t) dlsym(handle, "proxysql_mysql_auth_plugin_create");
+			char* dlsym_error = dlerror();
+			if (dlsym_error || !create_func) {
+				proxy_error("Auth plugin '%s' missing proxysql_mysql_auth_plugin_create\n",
+					plugin_path.c_str());
+				dlclose(handle);
+				continue;
+			}
+
+			// Create plugin instance
+			ProxySQL_Auth_Plugin* plugin = create_func();
+			if (!plugin) {
+				proxy_error("Failed to create auth plugin from '%s'\n", plugin_path.c_str());
+				dlclose(handle);
+				continue;
+			}
+
+			// Get plugin name
+			const char* name = plugin->name();
+			if (!name || strlen(name) == 0) {
+				proxy_error("Auth plugin '%s' returned empty name\n", plugin_path.c_str());
+				delete plugin;
+				dlclose(handle);
+				continue;
+			}
+
+			// Check for duplicate
+			if (GloAuthPlugins.find(name) != GloAuthPlugins.end()) {
+				proxy_error("Auth plugin '%s' already loaded, skipping '%s'\n",
+					name, plugin_path.c_str());
+				delete plugin;
+				dlclose(handle);
+				continue;
+			}
+
+			// Initialize the plugin
+			if (!plugin->init()) {
+				proxy_error("Failed to initialize auth plugin '%s'\n", name);
+				delete plugin;
+				dlclose(handle);
+				continue;
+			}
+
+			GloAuthPlugins[name] = plugin;
+			proxy_info("Loaded auth plugin: %s from %s\n", name, plugin_path.c_str());
+			plugin->print_version();
+		}
+	}
 }
 
 /**
diff --git a/test/tap/tests/run_mysql_authplugin_test.sh b/test/tap/tests/run_mysql_authplugin_test.sh
new file mode 100755
index 0000000000..1215c83ae3
--- /dev/null
+++ b/test/tap/tests/run_mysql_authplugin_test.sh
@@ -0,0 +1,122 @@
+#!/bin/bash
+#
+# Run auth plugin TAP tests locally
+#
+# Usage:
+#   ./run_auth_plugin_test.sh              # Run from inside dev container
+#   docker compose exec dev ./test/tap/tests/run_auth_plugin_test.sh  # Run from host
+#
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROXYSQL_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
+
+# Colors for output
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+NC='\033[0m' # No Color
+
+log_info() { echo -e "${GREEN}[INFO]${NC} $1"; }
+log_warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
+log_error() { echo -e "${RED}[ERROR]${NC} $1"; }
+
+# Configuration (can be overridden via environment)
+: "${TAP_MYSQLHOST:=mariadb}"
+: "${TAP_MYSQLPORT:=3306}"
+: "${TAP_MYSQLPASSWORD:=root}"
+: "${TAP_MYSQLUSERNAME:=root}"
+: "${TAP_HOST:=127.0.0.1}"
+: "${TAP_PORT:=6033}"
+: "${TAP_ADMINHOST:=127.0.0.1}"
+: "${TAP_ADMINPORT:=6032}"
+: "${TAP_ADMINUSERNAME:=admin}"
+: "${TAP_ADMINPASSWORD:=admin}"
+
+export TAP_MYSQLHOST TAP_MYSQLPORT TAP_MYSQLPASSWORD TAP_MYSQLUSERNAME
+export TAP_HOST TAP_PORT TAP_ADMINHOST TAP_ADMINPORT TAP_ADMINUSERNAME TAP_ADMINPASSWORD
+
+cleanup() {
+    log_info "Cleaning up..."
+    # Graceful shutdown via admin interface
+    if mysql -h"$TAP_ADMINHOST" -P"$TAP_ADMINPORT" -u"$TAP_ADMINUSERNAME" -p"$TAP_ADMINPASSWORD" \
+        -N -e "PROXYSQL SHUTDOWN" 2>/dev/null; then
+        sleep 2
+    fi
+    # Force kill if still running
+    pkill -9 proxysql 2>/dev/null || true
+}
+
+trap cleanup EXIT
+
+# Step 1: Build ProxySQL if needed
+if [[ ! -f "$PROXYSQL_ROOT/src/proxysql" ]]; then
+    log_info "Building ProxySQL..."
+    make -C "$PROXYSQL_ROOT" -j"$(nproc)"
+else
+    log_info "ProxySQL binary found, skipping build"
+fi
+
+# Step 2: Build auth plugin
+log_info "Building static auth plugin..."
+make -C "$PROXYSQL_ROOT/plugins/MySQL_AuthPlugin/static" clean all
+
+# Step 3: Compile test
+log_info "Compiling auth plugin test..."
+cd "$SCRIPT_DIR"
+
+# Compile TAP helper if needed
+if [[ ! -f tap.o ]]; then
+    g++ -c ../tap/tap.cpp -I"$PROXYSQL_ROOT/include" -o tap.o
+fi
+
+g++ -o test_mysql_authplugin-t test_mysql_authplugin-t.cpp tap.o \
+    -I../tap \
+    -I"$PROXYSQL_ROOT/deps/mariadb-client-library/mariadb_client/include" \
+    -L"$PROXYSQL_ROOT/deps/mariadb-client-library/mariadb_client/libmariadb" \
+    -std=c++17 -O2 \
+    -lmariadb -lpthread -lm -lz -lssl -lcrypto
+
+# Step 4: Start ProxySQL
+log_info "Starting ProxySQL with auth plugin..."
+pkill -9 proxysql 2>/dev/null || true
+rm -rf /tmp/proxysql
+mkdir -p /tmp/proxysql
+
+cd "$PROXYSQL_ROOT"
+./src/proxysql -f -c test_auth_plugin.cfg > /tmp/proxysql/proxysql.log 2>&1 &
+PROXYSQL_PID=$!
+
+# Wait for ProxySQL to start
+sleep 3
+
+if ! ps -p $PROXYSQL_PID > /dev/null 2>&1; then
+    log_error "ProxySQL failed to start. Log:"
+    cat /tmp/proxysql/proxysql.log
+    exit 1
+fi
+
+log_info "ProxySQL started (PID: $PROXYSQL_PID)"
+
+# Check if auth plugin loaded
+if ! mysql -h"$TAP_ADMINHOST" -P"$TAP_ADMINPORT" -u"$TAP_ADMINUSERNAME" -p"$TAP_ADMINPASSWORD" \
+    -N -e "SELECT 1" 2>/dev/null | grep -q 1; then
+    log_error "Cannot connect to ProxySQL admin interface"
+    exit 1
+fi
+
+# Step 5: Run test
+log_info "Running auth plugin test..."
+echo ""
+cd "$SCRIPT_DIR"
+./test_mysql_authplugin-t
+TEST_RC=$?
+
+echo ""
+if [[ $TEST_RC -eq 0 ]]; then
+    log_info "All tests passed!"
+else
+    log_error "Some tests failed (exit code: $TEST_RC)"
+fi
+
+exit $TEST_RC
diff --git a/test/tap/tests/test_mysql_authplugin-t.cpp b/test/tap/tests/test_mysql_authplugin-t.cpp
new file mode 100644
index 0000000000..aa754b0609
--- /dev/null
+++ b/test/tap/tests/test_mysql_authplugin-t.cpp
@@ -0,0 +1,263 @@
+/**
+ * @file test_auth_plugin-t.cpp
+ * @brief Tests for the per-user authentication plugin framework
+ * @details Tests the ProxySQL_Auth_Plugin interface with the static auth plugin:
+ *   - Plugin loading verification
+ *   - Authentication with correct password
+ *   - Authentication with wrong password (should fail)
+ *   - Backend username mapping via attributes
+ *   - Fallback to normal auth when no auth_plugin attribute
+ *
+ * Prerequisites:
+ *   - ProxySQL running with auth_plugins="/path/to/auth_static.so"
+ *   - MySQL/MariaDB backend accessible
+ *
+ * Environment variables:
+ *   TAP_ADMINUSERNAME, TAP_ADMINPASSWORD, TAP_ADMINPORT (default: admin/admin/6032)
+ *   TAP_HOST, TAP_PORT (default: 127.0.0.1/6033)
+ *   TAP_MYSQLUSERNAME, TAP_MYSQLPASSWORD, TAP_MYSQLPORT (default: root/root/3306)
+ */
+
+#include <cstring>
+#include <string>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+
+#include "mysql.h"
+#include "tap.h"
+
+using std::string;
+
+// Simple getenv with default
+const char* getenv_default(const char* name, const char* def) {
+    const char* val = getenv(name);
+    return val ? val : def;
+}
+
+int getenv_int(const char* name, int def) {
+    const char* val = getenv(name);
+    return val ? atoi(val) : def;
+}
+
+/**
+ * @brief Helper to execute admin query with error handling
+ */
+int admin_query(MYSQL* admin, const char* query) {
+    diag("Running: %s", query);
+    int rc = mysql_query(admin, query);
+    if (rc != 0) {
+        diag("Query failed: %s", mysql_error(admin));
+    }
+    return rc;
+}
+
+/**
+ * @brief Setup test users for auth plugin testing
+ */
+int setup_test_users(MYSQL* admin) {
+    // Clean up existing test users
+    admin_query(admin, "DELETE FROM mysql_users WHERE username IN ('auth_plugin_user', 'normal_user', 'backend_user')");
+
+    // Add backend-only user (frontend=0, backend=1) with password for connecting to MySQL
+    const char* backend_user_sql =
+        "INSERT INTO mysql_users (username, password, default_hostgroup, frontend, backend) "
+        "VALUES ('backend_user', 'backend_pass', 1, 0, 1)";
+    if (admin_query(admin, backend_user_sql) != 0) return -1;
+
+    // Add frontend user with static auth plugin (frontend=1, backend=0)
+    const char* auth_plugin_user_sql =
+        "INSERT INTO mysql_users (username, password, default_hostgroup, frontend, backend, attributes) "
+        "VALUES ('auth_plugin_user', '', 1, 1, 0, "
+        "'{\"auth_plugin\": \"static\", \"static_password\": \"plugin_secret\", \"backend_username\": \"backend_user\"}')";
+    if (admin_query(admin, auth_plugin_user_sql) != 0) return -1;
+
+    // Add normal user without auth_plugin (uses standard ProxySQL auth)
+    const char* normal_user_sql =
+        "INSERT INTO mysql_users (username, password, default_hostgroup, frontend, backend) "
+        "VALUES ('normal_user', 'normal_pass', 1, 1, 1)";
+    if (admin_query(admin, normal_user_sql) != 0) return -1;
+
+    // Load users to runtime
+    if (admin_query(admin, "LOAD MYSQL USERS TO RUNTIME") != 0) return -1;
+
+    return 0;
+}
+
+/**
+ * @brief Setup MySQL backend server
+ */
+int setup_backend_server(MYSQL* admin, const char* mysql_host, int mysql_port) {
+    char query[512];
+
+    // Clean and add backend server
+    admin_query(admin, "DELETE FROM mysql_servers WHERE hostgroup_id=1");
+    snprintf(query, sizeof(query),
+        "INSERT INTO mysql_servers (hostgroup_id, hostname, port) VALUES (1, '%s', %d)",
+        mysql_host, mysql_port);
+    if (admin_query(admin, query) != 0) return -1;
+    if (admin_query(admin, "LOAD MYSQL SERVERS TO RUNTIME") != 0) return -1;
+
+    return 0;
+}
+
+/**
+ * @brief Create test users on the MySQL backend
+ */
+int setup_mysql_backend_users(MYSQL* mysql_admin) {
+    // Create backend_user on MySQL
+    admin_query(mysql_admin, "DROP USER IF EXISTS 'backend_user'@'%'");
+    admin_query(mysql_admin, "CREATE USER 'backend_user'@'%' IDENTIFIED BY 'backend_pass'");
+    admin_query(mysql_admin, "GRANT ALL ON *.* TO 'backend_user'@'%'");
+
+    // Create normal_user on MySQL
+    admin_query(mysql_admin, "DROP USER IF EXISTS 'normal_user'@'%'");
+    admin_query(mysql_admin, "CREATE USER 'normal_user'@'%' IDENTIFIED BY 'normal_pass'");
+    admin_query(mysql_admin, "GRANT ALL ON *.* TO 'normal_user'@'%'");
+
+    admin_query(mysql_admin, "FLUSH PRIVILEGES");
+
+    return 0;
+}
+
+/**
+ * @brief Test connection attempt and return success/failure
+ */
+bool test_connection(const char* host, int port, const char* user, const char* pass, string& out_user) {
+    MYSQL* conn = mysql_init(NULL);
+    if (!conn) return false;
+
+    // Use mysql_clear_password for auth plugin users (sends cleartext)
+    mysql_options(conn, MYSQL_DEFAULT_AUTH, "mysql_clear_password");
+
+    // Enable cleartext plugin
+    my_bool enable_cleartext = 1;
+    mysql_options(conn, MYSQL_ENABLE_CLEARTEXT_PLUGIN, &enable_cleartext);
+
+    bool success = false;
+    if (mysql_real_connect(conn, host, user, pass, NULL, port, NULL, 0)) {
+        // Query current user to verify backend mapping
+        if (mysql_query(conn, "SELECT CURRENT_USER()") == 0) {
+            MYSQL_RES* res = mysql_store_result(conn);
+            if (res) {
+                MYSQL_ROW row = mysql_fetch_row(res);
+                if (row && row[0]) {
+                    out_user = row[0];
+                }
+                mysql_free_result(res);
+            }
+        }
+        success = true;
+    }
+
+    mysql_close(conn);
+    return success;
+}
+
+int main(int argc, char** argv) {
+    // Configuration from environment
+    const char* admin_host = getenv_default("TAP_ADMINHOST", "127.0.0.1");
+    int admin_port = getenv_int("TAP_ADMINPORT", 6032);
+    const char* admin_user = getenv_default("TAP_ADMINUSERNAME", "admin");
+    const char* admin_pass = getenv_default("TAP_ADMINPASSWORD", "admin");
+
+    const char* proxy_host = getenv_default("TAP_HOST", "127.0.0.1");
+    int proxy_port = getenv_int("TAP_PORT", 6033);
+
+    const char* mysql_host = getenv_default("TAP_MYSQLHOST", "127.0.0.1");
+    int mysql_port = getenv_int("TAP_MYSQLPORT", 3306);
+    const char* mysql_user = getenv_default("TAP_MYSQLUSERNAME", "root");
+    const char* mysql_pass = getenv_default("TAP_MYSQLPASSWORD", "root");
+
+    // Total number of tests
+    plan(6);
+
+    // Connect to ProxySQL admin
+    MYSQL* admin = mysql_init(NULL);
+    if (!mysql_real_connect(admin, admin_host, admin_user, admin_pass,
+                            NULL, admin_port, NULL, 0)) {
+        fprintf(stderr, "Failed to connect to ProxySQL admin: %s\n", mysql_error(admin));
+        return EXIT_FAILURE;
+    }
+
+    // Connect to MySQL backend to create test users
+    MYSQL* mysql_admin = mysql_init(NULL);
+    if (!mysql_real_connect(mysql_admin, mysql_host, mysql_user, mysql_pass,
+                            NULL, mysql_port, NULL, 0)) {
+        fprintf(stderr, "Failed to connect to MySQL backend: %s\n", mysql_error(mysql_admin));
+        diag("Skipping tests - cannot connect to MySQL backend at %s:%d", mysql_host, mysql_port);
+        skip(6, "MySQL backend not available");
+        mysql_close(admin);
+        return exit_status();
+    }
+
+    // Setup backend server in ProxySQL
+    if (setup_backend_server(admin, mysql_host, mysql_port) != 0) {
+        diag("Failed to setup backend server");
+        skip(6, "Backend server setup failed");
+        mysql_close(mysql_admin);
+        mysql_close(admin);
+        return exit_status();
+    }
+
+    // Setup users on MySQL backend
+    if (setup_mysql_backend_users(mysql_admin) != 0) {
+        diag("Failed to setup MySQL backend users");
+        skip(6, "MySQL user setup failed");
+        mysql_close(mysql_admin);
+        mysql_close(admin);
+        return exit_status();
+    }
+    mysql_close(mysql_admin);
+
+    // Setup test users in ProxySQL
+    if (setup_test_users(admin) != 0) {
+        diag("Failed to setup ProxySQL test users");
+        skip(6, "ProxySQL user setup failed");
+        mysql_close(admin);
+        return exit_status();
+    }
+
+    string backend_user;
+
+    // Test 1: Auth plugin user with correct password should succeed
+    diag("Testing auth_plugin_user with correct password...");
+    bool auth_success = test_connection(proxy_host, proxy_port, "auth_plugin_user", "plugin_secret", backend_user);
+    ok(auth_success, "Auth plugin user with correct password should connect successfully");
+
+    // Test 2: Backend username mapping should work
+    if (auth_success) {
+        ok(backend_user.find("backend_user") != string::npos,
+           "Backend username mapping: expected 'backend_user', got '%s'", backend_user.c_str());
+    } else {
+        ok(false, "Backend username mapping - skipped due to connection failure");
+    }
+
+    // Test 3: Auth plugin user with wrong password should fail
+    diag("Testing auth_plugin_user with wrong password...");
+    bool wrong_pass = test_connection(proxy_host, proxy_port, "auth_plugin_user", "wrong_password", backend_user);
+    ok(!wrong_pass, "Auth plugin user with wrong password should be rejected");
+
+    // Test 4: Normal user (no auth_plugin) with correct password should succeed
+    diag("Testing normal_user with correct password...");
+    bool normal_success = test_connection(proxy_host, proxy_port, "normal_user", "normal_pass", backend_user);
+    ok(normal_success, "Normal user with correct password should connect successfully");
+
+    // Test 5: Normal user with wrong password should fail
+    diag("Testing normal_user with wrong password...");
+    bool normal_wrong = test_connection(proxy_host, proxy_port, "normal_user", "wrong_pass", backend_user);
+    ok(!normal_wrong, "Normal user with wrong password should be rejected");
+
+    // Test 6: Non-existent user should fail
+    diag("Testing non-existent user...");
+    bool nonexistent = test_connection(proxy_host, proxy_port, "nonexistent_user", "any_pass", backend_user);
+    ok(!nonexistent, "Non-existent user should be rejected");
+
+    // Cleanup
+    admin_query(admin, "DELETE FROM mysql_users WHERE username IN ('auth_plugin_user', 'normal_user', 'backend_user')");
+    admin_query(admin, "LOAD MYSQL USERS TO RUNTIME");
+
+    mysql_close(admin);
+
+    return exit_status();
+}
diff --git a/test/tap/tests/unit-k8s_auth_plugin-t.cpp b/test/tap/tests/unit-k8s_auth_plugin-t.cpp
new file mode 100644
index 0000000000..7649bc6f00
--- /dev/null
+++ b/test/tap/tests/unit-k8s_auth_plugin-t.cpp
@@ -0,0 +1,285 @@
+/**
+ * @file unit-k8s_auth_plugin-t.cpp
+ * @brief Unit tests for K8s auth plugin using mock Kubernetes client
+ *
+ * Tests the K8s_Auth_Plugin class in isolation using MockKubernetesClient
+ * for dependency injection. No actual K8s cluster or ProxySQL required.
+ *
+ * Test cases:
+ *   1. Valid token with matching username
+ *   2. Valid token with mismatched username
+ *   3. Invalid/expired token
+ *   4. API error response
+ *   5. Empty token
+ *   6. Backend username mapping from attributes
+ *   7. NULL username parameter
+ *   8. NULL attributes parameter
+ *   9. Non-ServiceAccount token
+ *  10. Malformed attributes JSON
+ *  11. HTTP 500 error from TokenReview API
+ *  12. Request timeout error handling
+ *  13. Slow API response (simulates stuck server)
+ */
+
+#include <cstdlib>
+#include <cstring>
+#include <string>
+#include <memory>
+#include <chrono>
+
+#include "tap.h"
+
+// Include plugin class and mock client
+#include "../../../plugins/MySQL_AuthPlugin/k8s/k8s_auth_plugin.h"
+#include "../../../plugins/MySQL_AuthPlugin/k8s/mock_kubernetes_client.h"
+
+// Helper to free result strings
+void free_result(ProxySQL_Auth_Result& result) {
+    if (result.backend_username) free((void*)result.backend_username);
+    if (result.error_msg) free((void*)result.error_msg);
+}
+
+int main(int argc, char** argv) {
+    plan(24);
+
+    // Test 1: Valid token with matching username
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        mock->addValidToken("valid-token-1", "proxysql", "testuser");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "valid-token-1", attrs);
+
+        ok(result.success, "Test 1: Valid token with matching username should succeed");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 2: Valid token with mismatched username
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        mock->addValidToken("valid-token-2", "proxysql", "testuser");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/wronguser", "valid-token-2", attrs);
+
+        ok(!result.success, "Test 2: Mismatched username should fail");
+        ok(result.error_msg && strstr(result.error_msg, "username mismatch") != nullptr,
+           "Test 2: Error should mention username mismatch");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 3: Invalid/expired token
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        mock->addInvalidToken("expired-token");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "expired-token", attrs);
+
+        ok(!result.success, "Test 3: Invalid token should fail");
+        ok(result.error_msg && strstr(result.error_msg, "not authenticated") != nullptr,
+           "Test 3: Error should mention authentication failure");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 4: API error response
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        mock->addErrorToken("error-token", "connection timeout");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "error-token", attrs);
+
+        ok(!result.success, "Test 4: API error should fail");
+        ok(result.error_msg && strstr(result.error_msg, "connection timeout") != nullptr,
+           "Test 4: Error message should contain API error");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 5: Empty token
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "", attrs);
+
+        ok(!result.success, "Test 5: Empty token should fail");
+        ok(result.error_msg && strstr(result.error_msg, "empty token") != nullptr,
+           "Test 5: Error should mention empty token");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 6: Backend username mapping from attributes
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        mock->addValidToken("token-with-mapping", "app", "myservice");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s", "backend_username": "dbuser"})";
+        ProxySQL_Auth_Result result = plugin.validate("app/myservice", "token-with-mapping", attrs);
+
+        ok(result.success && result.backend_username &&
+           strcmp(result.backend_username, "dbuser") == 0,
+           "Test 6: Backend username mapping should work");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 7: NULL username parameter
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate(nullptr, "some-token", attrs);
+
+        ok(!result.success, "Test 7: NULL username should fail");
+        ok(result.error_msg && strstr(result.error_msg, "missing required parameters") != nullptr,
+           "Test 7: Error should mention missing parameters");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 8: NULL attributes parameter
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "some-token", nullptr);
+
+        ok(!result.success, "Test 8: NULL attributes should fail");
+        ok(result.error_msg && strstr(result.error_msg, "missing required parameters") != nullptr,
+           "Test 8: Error should mention missing parameters");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 9: Non-ServiceAccount token (user token without system:serviceaccount: prefix)
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        // Simulate a user token that authenticates but isn't a ServiceAccount
+        TokenReviewResult user_token_result;
+        user_token_result.success = true;
+        user_token_result.authenticated = true;
+        user_token_result.username = "user:admin";  // Not a ServiceAccount
+        mock->addTokenResponse("user-token", user_token_result);
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "user-token", attrs);
+
+        ok(!result.success, "Test 9: Non-ServiceAccount token should fail");
+        ok(result.error_msg && strstr(result.error_msg, "not a ServiceAccount token") != nullptr,
+           "Test 9: Error should mention ServiceAccount");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 10: Malformed attributes JSON
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        mock->addValidToken("valid-token", "proxysql", "testuser");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = "not valid json {{{";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "valid-token", attrs);
+
+        ok(!result.success, "Test 10: Malformed JSON should fail");
+        // JSON parse errors come from nlohmann::json
+        ok(result.error_msg != nullptr, "Test 10: Should have error message");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 11: HTTP 500 error from TokenReview API
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        // Simulate HTTP 500 error (matches K8sHttpClient error format)
+        mock->addErrorToken("token-500", "TokenReview API returned HTTP 500");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "token-500", attrs);
+
+        ok(!result.success, "Test 11: HTTP 500 should fail");
+        ok(result.error_msg && strstr(result.error_msg, "HTTP 500") != nullptr,
+           "Test 11: Error should mention HTTP 500");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 12: Request timeout error handling
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        // Simulate curl timeout error (matches K8sHttpClient error format)
+        mock->addErrorToken("token-timeout", "HTTP request failed: Timeout was reached");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "token-timeout", attrs);
+
+        ok(!result.success, "Test 12: Timeout error should fail");
+        ok(result.error_msg && strstr(result.error_msg, "Timeout") != nullptr,
+           "Test 12: Error should mention timeout");
+        free_result(result);
+        plugin.deinit();
+    }
+
+    // Test 13: Slow API response (simulates stuck server)
+    {
+        auto mock = std::make_unique<MockKubernetesClient>();
+        // Add a token that takes 100ms to respond
+        mock->addSlowToken("slow-token", 100, "proxysql", "testuser");
+
+        K8s_Auth_Plugin plugin(std::move(mock));
+        plugin.init();
+
+        const char* attrs = R"({"auth_plugin": "k8s"})";
+
+        auto start = std::chrono::steady_clock::now();
+        ProxySQL_Auth_Result result = plugin.validate("proxysql/testuser", "slow-token", attrs);
+        auto end = std::chrono::steady_clock::now();
+        auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
+
+        ok(result.success, "Test 13: Slow token should eventually succeed");
+        ok(elapsed_ms >= 100, "Test 13: Should have waited at least 100ms (actual: %ldms)", elapsed_ms);
+        free_result(result);
+        plugin.deinit();
+    }
+
+    return exit_status();
+}