[Feature] Implement role-based access control with ADMIN and READ_ONLY

amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java

@@ -21,11 +21,13 @@
 import org.apache.amoro.config.ConfigOption;
 import org.apache.amoro.config.ConfigOptions;
 import org.apache.amoro.server.authentication.DefaultPasswdAuthenticationProvider;
+import org.apache.amoro.server.authorization.Role;
 import org.apache.amoro.utils.MemorySize;
 
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 public class AmoroManagementConf {
 
@@ -53,6 +55,71 @@ public class AmoroManagementConf {
           .defaultValue("admin")
           .withDescription("The administrator password");
 
+  public static final ConfigOption<Boolean> AUTHORIZATION_ENABLED =
+      ConfigOptions.key("http-server.authorization.enabled")
+          .booleanType()
+          .defaultValue(false)
+          .withDescription("Whether to enable dashboard RBAC authorization.");
+
+  public static final ConfigOption<Role> AUTHORIZATION_DEFAULT_ROLE =
+      ConfigOptions.key("http-server.authorization.default-role")
+          .enumType(Role.class)
+          .defaultValue(Role.READ_ONLY)
+          .withDescription(
+              "Default role for authenticated users without an explicit role mapping.");
+
+  public static final ConfigOption<List<String>> AUTHORIZATION_ADMIN_USERS =
+      ConfigOptions.key("http-server.authorization.admin-users")
+          .stringType()
+          .asList()
+          .defaultValues()
+          .withDescription("Additional usernames that should always be treated as admin users.");
+
+  public static final ConfigOption<List<Map<String, String>>> AUTHORIZATION_USERS =
+      ConfigOptions.key("http-server.authorization.users")
+          .mapType()
+          .asList()
+          .noDefaultValue()
+          .withDescription("Local dashboard users with username/password/role entries.");
+
+  public static final ConfigOption<Boolean> AUTHORIZATION_LDAP_ROLE_MAPPING_ENABLED =
+      ConfigOptions.key("http-server.authorization.ldap-role-mapping.enabled")
+          .booleanType()
+          .defaultValue(false)
+          .withDescription("Whether to resolve dashboard roles from LDAP group membership.");
+
+  public static final ConfigOption<String> AUTHORIZATION_LDAP_ROLE_MAPPING_ADMIN_GROUP_DN =
+      ConfigOptions.key("http-server.authorization.ldap-role-mapping.admin-group-dn")
+          .stringType()
+          .noDefaultValue()
+          .withDescription(
+              "Full DN of the LDAP admin group, e.g. CN=amoro-admins,OU=Groups,DC=example,DC=com.");
+
+  public static final ConfigOption<String> AUTHORIZATION_LDAP_ROLE_MAPPING_GROUP_MEMBER_ATTRIBUTE =
+      ConfigOptions.key("http-server.authorization.ldap-role-mapping.group-member-attribute")
+          .stringType()
+          .defaultValue("member")
+          .withDescription("LDAP group attribute that stores member references.");
+
+  public static final ConfigOption<String> AUTHORIZATION_LDAP_ROLE_MAPPING_USER_DN_PATTERN =
+      ConfigOptions.key("http-server.authorization.ldap-role-mapping.user-dn-pattern")
+          .stringType()
+          .noDefaultValue()
+          .withDescription(
+              "LDAP user DN pattern used to match group members. Use {0} as the username placeholder.");
+
+  public static final ConfigOption<String> AUTHORIZATION_LDAP_ROLE_MAPPING_BIND_DN =
+      ConfigOptions.key("http-server.authorization.ldap-role-mapping.bind-dn")
+          .stringType()
+          .defaultValue("")
+          .withDescription("Optional LDAP bind DN used when querying role-mapping groups.");
+
+  public static final ConfigOption<String> AUTHORIZATION_LDAP_ROLE_MAPPING_BIND_PASSWORD =
+      ConfigOptions.key("http-server.authorization.ldap-role-mapping.bind-password")
+          .stringType()
+          .defaultValue("")
+          .withDescription("Optional LDAP bind password used when querying role-mapping groups.");
+
   /** Enable master & slave mode, which supports horizontal scaling of AMS. */
   public static final ConfigOption<Boolean> USE_MASTER_SLAVE_MODE =
       ConfigOptions.key("use-master-slave-mode")

amoro-ams/src/main/java/org/apache/amoro/server/authentication/DefaultPasswdAuthenticationProvider.java

@@ -24,22 +24,70 @@
 import org.apache.amoro.config.Configurations;
 import org.apache.amoro.exception.SignatureCheckException;
 import org.apache.amoro.server.AmoroManagementConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 public class DefaultPasswdAuthenticationProvider implements PasswdAuthenticationProvider {
-  private String basicAuthUser;
-  private String basicAuthPassword;
+  private static final Logger LOG =
+      LoggerFactory.getLogger(DefaultPasswdAuthenticationProvider.class);
+
+  private final String basicAuthUser;
+  private final String basicAuthPassword;
+  private final Map<String, String> localUsers;
 
   public DefaultPasswdAuthenticationProvider(Configurations conf) {
     this.basicAuthUser = conf.get(AmoroManagementConf.ADMIN_USERNAME);
     this.basicAuthPassword = conf.get(AmoroManagementConf.ADMIN_PASSWORD);
+    this.localUsers = loadLocalUsers(conf);
   }
 
   @Override
   public BasicPrincipal authenticate(PasswordCredential credential) throws SignatureCheckException {
+    String localPassword = localUsers.get(credential.username());
+    if (localPassword != null) {
+      if (localPassword.equals(credential.password())) {
+        return new BasicPrincipal(credential.username());
+      }
+      throw new SignatureCheckException("Invalid password for user: " + credential.username());
+    }
+
     if (!(basicAuthUser.equals(credential.username())
         && basicAuthPassword.equals(credential.password()))) {
       throw new SignatureCheckException("Failed to authenticate via basic authentication");
     }
     return new BasicPrincipal(credential.username());
   }
+
+  private static Map<String, String> loadLocalUsers(Configurations conf) {
+    if (!conf.get(AmoroManagementConf.AUTHORIZATION_ENABLED)) {
+      return Collections.emptyMap();
+    }
+
+    List<Map<String, String>> users =
+        conf.getOptional(AmoroManagementConf.AUTHORIZATION_USERS).orElse(Collections.emptyList());
+    return users.stream()
+        .filter(DefaultPasswdAuthenticationProvider::hasRequiredLocalAuthFields)
+        .collect(
+            Collectors.toMap(
+                user -> String.valueOf(user.get("username")),
+                user -> String.valueOf(user.get("password")),
+                (existing, replacement) -> {
+                  LOG.warn(
+                      "Duplicate authorization.users entry for password auth, keeping last user definition");
+                  return replacement;
+                }));
+  }
+
+  private static boolean hasRequiredLocalAuthFields(Map<String, String> user) {
+    if (user.get("username") == null || user.get("password") == null) {
+      LOG.warn("Ignore invalid authorization.users entry for password auth: {}", user);
+      return false;
+    }
+    return true;
+  }
 }

amoro-ams/src/main/java/org/apache/amoro/server/authentication/HttpAuthenticationFactory.java

@@ -56,8 +56,15 @@ private static <T> T createAuthenticationProvider(
           .impl(className)
           .<T>buildChecked()
           .newInstance(conf);
+    } catch (NoSuchMethodException e) {
+      throw new IllegalStateException(
+          className
+              + " must implement "
+              + expected.getName()
+              + " and provide a public constructor (Configurations) or no-arg constructor",
+          e);
     } catch (Exception e) {
-      throw new IllegalStateException(className + " must extend of " + expected.getName());
+      throw new IllegalStateException("Failed to create " + className + ": " + e.getMessage(), e);
     }
   }
 

amoro-ams/src/main/java/org/apache/amoro/server/authentication/LdapPasswdAuthenticationProvider.java

@@ -54,12 +54,14 @@ public LdapPasswdAuthenticationProvider(Configurations conf) {
 
   @Override
   public BasicPrincipal authenticate(PasswordCredential credential) throws SignatureCheckException {
+    String username = normalizeUsername(credential.username());
     Hashtable<String, Object> env = new Hashtable<String, Object>();
     env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.ldap.LdapCtxFactory");
     env.put(Context.PROVIDER_URL, ldapUrl);
     env.put(Context.SECURITY_AUTHENTICATION, "simple");
     env.put(Context.SECURITY_CREDENTIALS, credential.password());
-    env.put(Context.SECURITY_PRINCIPAL, formatter.format(new String[] {credential.username()}));
+    env.put(Context.SECURITY_PRINCIPAL, formatter.format(new String[] {username}));
+    env.put(Context.REFERRAL, "follow");
 
     InitialDirContext initialLdapContext = null;
     try {
@@ -75,6 +77,17 @@ public BasicPrincipal authenticate(PasswordCredential credential) throws Signatu
         }
       }
     }
-    return new BasicPrincipal(credential.username());
+    return new BasicPrincipal(username);
+  }
+
+  /**
+   * Strip email domain suffix if present so that "xuba@cisco.com" becomes "xuba". The LDAP
+   * user-pattern template expects a plain username, not an email address.
+   */
+  private static String normalizeUsername(String username) {
+    if (username != null && username.contains("@")) {
+      return username.substring(0, username.indexOf('@'));
+    }
+    return username;
   }
 }