[KYUUBI #6750] [REST] Using ForbiddenException instead of NotAllowedException

# 🔍 Description
## Issue References 🔗

Seems NotAllowedException is used for method not allowed, and currently, we use false constructor, the error message we expected would not be return to client end.

It only told:
```
{"message":"HTTP 405 Method Not Allowed"}
```
Because the message we used to build the NotAllowedException was treated as `allowed` method, not as `message`.

![image](https://github.com/user-attachments/assets/3199f20c-6148-4e6a-9183-7a0843913d8d)

## Describe Your Solution 🔧

We should use the ForbidenException instead, and then the error message we excepted can be visible in client end.

https://github.com/apache/kyuubi/blob/85dd5a52efa790195ed46b713c53938ffd31c5d9/kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/api.scala#L47-L51

## Types of changes 🔖

- [ ] Bugfix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)

## Test Plan 🧪

#### Behavior Without This Pull Request ⚰️

#### Behavior With This Pull Request 🎉

#### Related Unit Tests

<img width="913" alt="image" src="https://github.com/user-attachments/assets/6c4e836d-a47a-485d-85a3-fd3a35a9e425">

---

# Checklist 📝

- [x] This patch was not authored or co-authored using [Generative Tooling](https://www.apache.org/legal/generative-tooling.html)

**Be nice. Be informative.**

Closes #6750 from turboFei/not_allowed_exception.

Closes #6750

4dd6fc1 [Wang, Fei] Using ForbiddenException instead of NotAllowedException

Authored-by: Wang, Fei <[email protected]>
Signed-off-by: Bowen Liang <[email protected]>
(cherry picked from commit cefb98d)
Signed-off-by: Bowen Liang <[email protected]>

Author: turboFei

kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala

@@ -248,7 +248,7 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
     } catch {
       case t: Throwable => throw new WebApplicationException(
           t.getMessage,
-          Status.METHOD_NOT_ALLOWED)
+          Status.FORBIDDEN)
     }
   }
 

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/AdminResource.scala

@@ -57,7 +57,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Receive refresh Kyuubi server hadoop conf request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to refresh the Kyuubi server hadoop conf")
     }
     info(s"Reloading the Kyuubi server hadoop conf")
@@ -76,7 +76,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Receive refresh user defaults conf request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to refresh the user defaults conf")
     }
     info(s"Reloading user defaults conf")
@@ -95,7 +95,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Receive refresh kubernetes conf request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to refresh the kubernetes conf")
     }
     info(s"Reloading kubernetes conf")
@@ -114,7 +114,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Receive refresh unlimited users request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to refresh the unlimited users")
     }
     info(s"Reloading unlimited users")
@@ -133,7 +133,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Receive refresh deny users request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to refresh the deny users")
     }
     info(s"Reloading deny users")
@@ -152,7 +152,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Receive refresh deny ips request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to refresh the deny ips")
     }
     info(s"Reloading deny ips")
@@ -175,7 +175,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Received listing all live sessions request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to list all live sessions")
     }
     var sessions = fe.be.sessionManager.allSessions()
@@ -201,7 +201,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Received closing a session request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to close the session $sessionHandleStr")
     }
     fe.be.closeSession(SessionHandle.fromUUID(sessionHandleStr))
@@ -226,7 +226,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Received listing all of the active operations request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to list all the operations")
     }
     var operations = fe.be.sessionManager.operationManager.allOperations()
@@ -258,7 +258,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Received close an operation request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to close the operation $operationHandleStr")
     }
     val operationHandle = OperationHandle(operationHandleStr)
@@ -371,7 +371,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Received list all live kyuubi servers request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to list all live kyuubi servers")
     }
     val kyuubiConf = fe.getConf
@@ -442,7 +442,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
     val ipAddress = fe.getIpAddress
     info(s"Received counting batches request from $userName/$ipAddress")
     if (!fe.isAdministrator(userName)) {
-      throw new NotAllowedException(
+      throw new ForbiddenException(
         s"$userName is not allowed to count the batches")
     }
     val batchCount = batchService

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/AdminResourceSuite.scala

@@ -90,7 +90,7 @@ class AdminResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
       .request()
       .header(AUTHORIZATION_HEADER, HttpAuthUtils.basicAuthorizationHeader("admin002"))
       .post(null)
-    assert(response.getStatus === 405)
+    assert(response.getStatus === 403)
   }
 
   test("refresh user defaults config of the kyuubi server") {
@@ -494,19 +494,19 @@ class AdminResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
 
       // use proxyUser
       val deleteEngineResponse1 = runDeleteEngine(Option(normalUser), None)
-      assert(deleteEngineResponse1.getStatus === 405)
+      assert(deleteEngineResponse1.getStatus === 403)
       val errorMessage = s"Failed to validate proxy privilege of anonymous for $normalUser"
       assert(deleteEngineResponse1.readEntity(classOf[String]).contains(errorMessage))
 
       // it should be the same behavior as hive.server2.proxy.user
       val deleteEngineResponse2 = runDeleteEngine(None, Option(normalUser))
-      assert(deleteEngineResponse2.getStatus === 405)
+      assert(deleteEngineResponse2.getStatus === 403)
       assert(deleteEngineResponse2.readEntity(classOf[String]).contains(errorMessage))
 
       // when both set, proxyUser takes precedence
       val deleteEngineResponse3 =
         runDeleteEngine(Option(normalUser), Option(s"${normalUser}HiveServer2"))
-      assert(deleteEngineResponse3.getStatus === 405)
+      assert(deleteEngineResponse3.getStatus === 403)
       assert(deleteEngineResponse3.readEntity(classOf[String]).contains(errorMessage))
     }
   }
@@ -745,19 +745,19 @@ class AdminResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
 
       // use proxyUser
       val listEngineResponse1 = runListEngine(Option(normalUser), None)
-      assert(listEngineResponse1.getStatus === 405)
+      assert(listEngineResponse1.getStatus === 403)
       val errorMessage = s"Failed to validate proxy privilege of anonymous for $normalUser"
       assert(listEngineResponse1.readEntity(classOf[String]).contains(errorMessage))
 
       // it should be the same behavior as hive.server2.proxy.user
       val listEngineResponse2 = runListEngine(None, Option(normalUser))
-      assert(listEngineResponse2.getStatus === 405)
+      assert(listEngineResponse2.getStatus === 403)
       assert(listEngineResponse2.readEntity(classOf[String]).contains(errorMessage))
 
       // when both set, proxyUser takes precedence
       val listEngineResponse3 =
         runListEngine(Option(normalUser), Option(s"${normalUser}HiveServer2"))
-      assert(listEngineResponse3.getStatus === 405)
+      assert(listEngineResponse3.getStatus === 403)
       assert(listEngineResponse3.readEntity(classOf[String]).contains(errorMessage))
     }
   }

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala

@@ -136,7 +136,7 @@ abstract class BatchesResourceSuiteBase extends KyuubiFunSuite
       .request(MediaType.APPLICATION_JSON_TYPE)
       .header(AUTHORIZATION_HEADER, basicAuthorizationHeader("anonymous"))
       .post(Entity.entity(proxyUserRequest, MediaType.APPLICATION_JSON_TYPE))
-    assert(proxyUserResponse.getStatus === 405)
+    assert(proxyUserResponse.getStatus === 403)
     var errorMessage = "Failed to validate proxy privilege of anonymous for root"
     assert(proxyUserResponse.readEntity(classOf[String]).contains(errorMessage))
 
@@ -211,7 +211,7 @@ abstract class BatchesResourceSuiteBase extends KyuubiFunSuite
       .request(MediaType.APPLICATION_JSON_TYPE)
       .header(AUTHORIZATION_HEADER, basicAuthorizationHeader(batch.getId))
       .delete()
-    assert(deleteBatchResponse.getStatus === 405)
+    assert(deleteBatchResponse.getStatus === 403)
     errorMessage = s"Failed to validate proxy privilege of ${batch.getId} for anonymous"
     assert(deleteBatchResponse.readEntity(classOf[String]).contains(errorMessage))
 
@@ -872,19 +872,19 @@ abstract class BatchesResourceSuiteBase extends KyuubiFunSuite
 
     // use kyuubi.session.proxy.user
     val proxyUserResponse1 = runOpenBatchExecutor(Option(normalUser), None)
-    assert(proxyUserResponse1.getStatus === 405)
+    assert(proxyUserResponse1.getStatus === 403)
     val errorMessage = s"Failed to validate proxy privilege of anonymous for $normalUser"
     assert(proxyUserResponse1.readEntity(classOf[String]).contains(errorMessage))
 
     // it should be the same behavior as hive.server2.proxy.user
     val proxyUserResponse2 = runOpenBatchExecutor(None, Option(normalUser))
-    assert(proxyUserResponse2.getStatus === 405)
+    assert(proxyUserResponse2.getStatus === 403)
     assert(proxyUserResponse2.readEntity(classOf[String]).contains(errorMessage))
 
     // when both set, kyuubi.session.proxy.user takes precedence
     val proxyUserResponse3 =
       runOpenBatchExecutor(Option(normalUser), Option(s"${normalUser}HiveServer2"))
-    assert(proxyUserResponse3.getStatus === 405)
+    assert(proxyUserResponse3.getStatus === 403)
     assert(proxyUserResponse3.readEntity(classOf[String]).contains(errorMessage))
   }
 }