KIKIMR-20378: Enable IAM BulkAuthorization (#911)

* Bulk authorization

* unit tests

* small fixes

* Add result filter ALL_FAILED

16 files changed, 844 insertions, 79 deletions

diff --git a/ydb/core/protos/feature_flags.proto b/ydb/core/protos/feature_flags.proto
index 79a55c2189b..7841c140320 100644
--- a/ydb/core/protos/feature_flags.proto
+++ b/ydb/core/protos/feature_flags.proto
@@ -126,4 +126,5 @@ message TFeatureFlags {
     optional bool UseVDisksBalancing = 111 [default = false];
     optional bool EnableViews = 112 [default = false];
     optional bool EnableServerlessExclusiveDynamicNodes = 113 [default = false];
+    optional bool EnableAccessServiceBulkAuthorization = 114 [default = false];
 }
diff --git a/ydb/core/security/ticket_parser_impl.h b/ydb/core/security/ticket_parser_impl.h
index 2dbe4b653e9..18aa1aa0dbf 100644
--- a/ydb/core/security/ticket_parser_impl.h
+++ b/ydb/core/security/ticket_parser_impl.h
@@ -48,9 +48,16 @@ private:
     };
 
     struct TPermissionRecord {
+        enum class TTypeCase {
+            TYPE_NOT_SET,
+            USER_ACCOUNT_TYPE,
+            SERVICE_ACCOUNT_TYPE,
+            ANONYMOUS_ACCOUNT_TYPE,
+        };
+
         TString Subject;
         bool Required = false;
-        yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase SubjectType;
+        TTypeCase SubjectType;
         TEvTicketParser::TError Error;
         TStackVec<std::pair<TString, TString>> Attributes;
 
@@ -78,6 +85,7 @@ private:
 
     using TEvAccessServiceAuthenticateRequest = TEvRequestWithKey<NCloud::TEvAccessService::TEvAuthenticateRequest>;
     using TEvAccessServiceAuthorizeRequest = TEvRequestWithKey<NCloud::TEvAccessService::TEvAuthorizeRequest>;
+    using TEvAccessServiceBulkAuthorizeRequest = TEvRequestWithKey<NCloud::TEvAccessService::TEvBulkAuthorizeRequest>;
     using TEvAccessServiceGetUserAccountRequest = TEvRequestWithKey<NCloud::TEvUserAccountService::TEvGetUserAccountRequest>;
     using TEvAccessServiceGetServiceAccountRequest = TEvRequestWithKey<NCloud::TEvServiceAccountService::TEvGetServiceAccountRequest>;
 
@@ -262,7 +270,8 @@ private:
     TDuration LifeTime = TDuration::Hours(1); // for how long ticket will remain in the cache after last access
     TDuration AsSignatureExpireTime = TDuration::Minutes(1);
 
-    TActorId AccessServiceValidator;
+    TActorId AccessServiceValidatorV1;
+    TActorId AccessServiceValidatorV2;
     TActorId UserAccountService;
     TActorId ServiceAccountService;
     TString UserAccountDomain;
@@ -350,42 +359,68 @@ private:
         return request;
     }
 
-    template <typename TTokenRecord>
-    void RequestAccessServiceAuthorization(const TString& key, TTokenRecord& record) const {
-        for (const auto& [perm, permRecord] : record.Permissions) {
-            const TString& permission(perm);
-            BLOG_TRACE("Ticket " << record.GetMaskedTicket() << " asking for AccessServiceAuthorization(" << permission << ")");
+    template <typename TTokenRecord, typename TPathsContainerPtr>
+    void addResourcePaths(const TTokenRecord& record, const TString& permission, TPathsContainerPtr pathsContainer) const {
+        auto addResourcePath = [&pathsContainer] (const TString& id, const TString& type) {
+            auto resourcePath = pathsContainer->add_resource_path();
+            resourcePath->set_id(id);
+            resourcePath->set_type(type);
+        };
 
-            auto request = CreateAccessServiceRequest<TEvAccessServiceAuthorizeRequest>(key, record);
+        if (const auto databaseId = record.GetAttributeValue(permission, "database_id"); databaseId) {
+            addResourcePath(databaseId, "ydb.database");
+        } else if (const auto serviceAccountId = record.GetAttributeValue(permission, "service_account_id"); serviceAccountId) {
+            addResourcePath(serviceAccountId, "iam.serviceAccount");
+        }
 
-            auto addResourcePath = [&request] (const TString& id, const TString& type) {
-                auto* resourcePath = request->Request.add_resource_path();
-                resourcePath->set_id(id);
-                resourcePath->set_type(type);
-            };
+        if (const auto folderId = record.GetAttributeValue(permission, "folder_id"); folderId) {
+            addResourcePath(folderId, "resource-manager.folder");
+        }
 
-            request->Request.set_permission(permission);
+        if (const auto cloudId = record.GetAttributeValue(permission, "cloud_id"); cloudId) {
+            addResourcePath(cloudId, "resource-manager.cloud");
+        }
 
-            if (const auto databaseId = record.GetAttributeValue(permission, "database_id"); databaseId) {
-                addResourcePath(databaseId, "ydb.database");
-            } else if (const auto serviceAccountId = record.GetAttributeValue(permission, "service_account_id"); serviceAccountId) {
-                addResourcePath(serviceAccountId, "iam.serviceAccount");
-            }
+        if (const TString gizmoId = record.GetAttributeValue(permission, "gizmo_id"); gizmoId) {
+            addResourcePath(gizmoId, "iam.gizmo");
+        }
+    }
 
-            if (const auto folderId = record.GetAttributeValue(permission, "folder_id"); folderId) {
-                addResourcePath(folderId, "resource-manager.folder");
-            }
+    template <typename TTokenRecord>
+    void AccessServiceAuthorize(const TString& key, TTokenRecord& record) const {
+        for (const auto& [permissionName, permissionRecord] : record.Permissions) {
+            BLOG_TRACE("Ticket " << record.GetMaskedTicket() << " asking for AccessServiceAuthorization(" << permissionName << ")");
 
-            if (const auto cloudId = record.GetAttributeValue(permission, "cloud_id"); cloudId) {
-                addResourcePath(cloudId, "resource-manager.cloud");
-            }
+            auto request = CreateAccessServiceRequest<TEvAccessServiceAuthorizeRequest>(key, record);
+            request->Request.set_permission(permissionName);
+            addResourcePaths(record, permissionName, &request->Request);
+            record.ResponsesLeft++;
+            Send(AccessServiceValidatorV1, request.Release());
+        }
+    }
 
-            if (const TString gizmoId = record.GetAttributeValue(permission, "gizmo_id"); gizmoId) {
-                addResourcePath(gizmoId, "iam.gizmo");
-            }
+    template <typename TTokenRecord>
+    void AccessServiceBulkAuthorize(const TString& key, TTokenRecord& record) const {
+        auto request = CreateAccessServiceRequest<TEvAccessServiceBulkAuthorizeRequest>(key, record);
+        TStringBuilder requestForPermissions;
+        for (const auto& [permissionName, permissionRecord] : record.Permissions) {
+            auto action = request->Request.mutable_actions()->add_items();
+            addResourcePaths(record, permissionName, action);
+            action->set_permission(permissionName);
+            requestForPermissions << " " << permissionName;
+        }
+        request->Request.set_result_filter(yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest::ALL_FAILED);
+        BLOG_TRACE("Ticket " << record.GetMaskedTicket() << " asking for AccessServiceBulkAuthorization(" << requestForPermissions << ")");
+        record.ResponsesLeft++;
+        Send(AccessServiceValidatorV2, request.Release());
+    }
 
-            record.ResponsesLeft++;
-            Send(AccessServiceValidator, request.Release());
+    template <typename TTokenRecord>
+    void RequestAccessServiceAuthorization(const TString& key, TTokenRecord& record) const {
+        if (AppData()->FeatureFlags.GetEnableAccessServiceBulkAuthorization()) {
+            AccessServiceBulkAuthorize(key, record);
+        } else {
+            AccessServiceAuthorize(key, record);
         }
     }
 
@@ -396,23 +431,38 @@ private:
         auto request = CreateAccessServiceRequest<TEvAccessServiceAuthenticateRequest>(key, record);
 
         record.ResponsesLeft++;
-        Send(AccessServiceValidator, request.Release());
+        Send(AccessServiceValidatorV1, request.Release());
     }
 
 
-    TString GetSubjectName(const yandex::cloud::priv::servicecontrol::v1::Subject& subject) {
+    template <typename TSubject>
+    TString GetSubjectName(const TSubject& subject) {
         switch (subject.type_case()) {
-        case yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase::kUserAccount:
+        case TSubject::TypeCase::kUserAccount:
             return subject.user_account().id() + "@" + AccessServiceDomain;
-        case yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase::kServiceAccount:
+        case TSubject::TypeCase::kServiceAccount:
             return subject.service_account().id() + "@" + AccessServiceDomain;
-        case yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase::kAnonymousAccount:
+        case TSubject::TypeCase::kAnonymousAccount:
             return "anonymous" "@" + AccessServiceDomain;
         default:
             return "Unknown subject type";
         }
     }
 
+    template <typename TSubjectType>
+    TPermissionRecord::TTypeCase ConvertSubjectType(const TSubjectType& type) {
+        switch (type) {
+        case TSubjectType::kUserAccount:
+            return TPermissionRecord::TTypeCase::USER_ACCOUNT_TYPE;
+        case TSubjectType::kServiceAccount:
+            return TPermissionRecord::TTypeCase::SERVICE_ACCOUNT_TYPE;
+        case TSubjectType::kAnonymousAccount:
+            return TPermissionRecord::TTypeCase::ANONYMOUS_ACCOUNT_TYPE;
+        default:
+            return TPermissionRecord::TTypeCase::TYPE_NOT_SET;
+        }
+    }
+
     template <typename TTokenRecord>
     bool CanInitBuiltinToken(const TString& key, TTokenRecord& record) {
         if (record.TokenType == TDerived::ETokenType::Unknown || record.TokenType == TDerived::ETokenType::Builtin) {
@@ -741,6 +791,140 @@ private:
         }
     }
 
+    template <typename TTokenRecord>
+    void SetAccessServiceBulkAuthorizeError(const TString& key, TTokenRecord& record, const TString& errorMessage, bool isRetryableError) {
+        for (auto& [permissionName, permissionRecord] : record.Permissions) {
+            permissionRecord.Subject.clear();
+            permissionRecord.Error = {.Message = errorMessage, .Retryable = isRetryableError};
+            BLOG_TRACE("Ticket " << record.GetMaskedTicket()
+                                << " permission " << permissionName
+                                << " now has a " << (isRetryableError ? "retryable" : "permanent")  << " error \"" << errorMessage << "\""
+                                << " retryable: " << isRetryableError);
+        }
+        SetError(key, record, {.Message = errorMessage, .Retryable = isRetryableError});
+    }
+
+    void Handle(NCloud::TEvAccessService::TEvBulkAuthorizeResponse::TPtr& ev) {
+        auto getResourcePathIdForRequiredPermissions = [] (const ::yandex::cloud::priv::accessservice::v2::Resource& resource_path) -> TString {
+            if (resource_path.type() == "resource-manager.folder") {
+                return " folder_id " + resource_path.id();
+            }
+            if (resource_path.type() == "resource-manager.cloud") {
+                return " cloud_id " + resource_path.id();
+            }
+            if (resource_path.type() == "iam.serviceAccount") {
+                return " service_account_id " + resource_path.id();
+            }
+            return "";
+        };
+
+        NCloud::TEvAccessService::TEvBulkAuthorizeResponse* response = ev->Get();
+        TEvAccessServiceBulkAuthorizeRequest* request = response->Request->Get<TEvAccessServiceBulkAuthorizeRequest>();
+        const TString& key(request->Key);
+        auto& userTokens = GetDerived()->GetUserTokens();
+        auto itToken = userTokens.find(key);
+        if (itToken == userTokens.end()) {
+            BLOG_ERROR("Ticket(key) "
+                        << MaskTicket(key)
+                        << " has expired during permission check");
+        } else {
+            auto& record = itToken->second;
+            --record.ResponsesLeft;
+            auto& examinedPermissions = record.Permissions;
+            if (response->Status.Ok()) {
+                if (response->Response.has_unauthenticated_error()) {
+                    SetAccessServiceBulkAuthorizeError(key, record, response->Response.unauthenticated_error().message(), false);
+                } else {
+                    const auto& subject = response->Response.subject();
+                    const TString subjectName = GetSubjectName(subject);
+                    const auto& subjectType = ConvertSubjectType(subject.type_case());
+                    for (auto& [permissionName, permissionRecord] : examinedPermissions) {
+                        permissionRecord.Subject = subjectName;
+                        permissionRecord.SubjectType = subjectType;
+                        permissionRecord.Error.clear();
+                    }
+                    size_t permissionDeniedCount = 0;
+                    bool hasRequiredPermissionFailed = false;
+                    std::vector<typename THashMap<TString, TPermissionRecord>::iterator> requiredPermissions;
+                    TString permissionDeniedError;
+                    const auto& results = response->Response.results();
+                    for (const auto& result : results.items()) {
+                        auto permissionDeniedIt = examinedPermissions.find(result.permission());
+                        if (permissionDeniedIt != examinedPermissions.end()) {
+                            permissionDeniedCount++;
+                            auto& permissionDeniedRecord = permissionDeniedIt->second;
+                            permissionDeniedRecord.Subject.clear();
+                            BLOG_TRACE("Ticket " << record.GetMaskedTicket() << " permission " << result.permission() << " access denied for subject \"" << subjectName << "\"");
+                            TStringBuilder errorMessage;
+                            if (permissionDeniedRecord.IsRequired()) {
+                                hasRequiredPermissionFailed = true;
+                                errorMessage << permissionDeniedIt->first << " for";
+                                for (const auto& resourcePath : result.resource_path()) {
+                                    errorMessage << getResourcePathIdForRequiredPermissions(resourcePath);
+                                }
+                                errorMessage << " - ";
+                                requiredPermissions.push_back(permissionDeniedIt);
+                            }
+                            permissionDeniedError = result.permission_denied_error().message();;
+                            errorMessage << permissionDeniedError;
+                            permissionDeniedRecord.Error = {.Message = errorMessage, .Retryable = false};
+                        } else {
+                            BLOG_W("Received response for unknown permission " << result.permission() << " for ticket " << record.GetMaskedTicket());
+                        }
+                    }
+                    if (permissionDeniedCount < examinedPermissions.size() && !hasRequiredPermissionFailed) {
+                        record.TokenType = request->Request.has_api_key() ? TDerived::ETokenType::ApiKey : TDerived::ETokenType::AccessService;
+                        switch (subjectType) {
+                        case TPermissionRecord::TTypeCase::USER_ACCOUNT_TYPE:
+                            if (UserAccountService) {
+                                BLOG_TRACE("Ticket " << record.GetMaskedTicket()
+                                            << " asking for UserAccount(" << subjectName << ")");
+                                THolder<TEvAccessServiceGetUserAccountRequest> request = MakeHolder<TEvAccessServiceGetUserAccountRequest>(key);
+                                request->Token = record.Ticket;
+                                request->Request.set_user_account_id(subject.user_account().id());
+                                record.ResponsesLeft++;
+                                Send(UserAccountService, request.Release());
+                                return;
+                            }
+                            break;
+                        case TPermissionRecord::TTypeCase::SERVICE_ACCOUNT_TYPE:
+                            if (ServiceAccountService) {
+                                BLOG_TRACE("Ticket " << record.GetMaskedTicket()
+                                            << " asking for ServiceAccount(" << subjectName << ")");
+                                THolder<TEvAccessServiceGetServiceAccountRequest> request = MakeHolder<TEvAccessServiceGetServiceAccountRequest>(key);
+                                request->Token = record.Ticket;
+                                request->Request.set_service_account_id(subject.service_account().id());
+                                record.ResponsesLeft++;
+                                Send(ServiceAccountService, request.Release());
+                                return;
+                            }
+                            break;
+                        default:
+                            break;
+                        }
+                        SetToken(request->Key, record, new NACLib::TUserToken(record.Ticket, subjectName, {}));
+                    } else {
+                        if (hasRequiredPermissionFailed) {
+                            TStringBuilder errorMessage;
+                            auto it = requiredPermissions.cbegin();
+                            errorMessage << (*it)->second.Error.Message;
+                            ++it;
+                            for (; it != requiredPermissions.cend(); ++it) {
+                                errorMessage << ", " << (*it)->second.Error.Message;
+                            }
+                            SetError(key, record, {.Message = errorMessage, .Retryable = false});
+                        } else {
+                            SetError(key, record, {.Message = permissionDeniedError, .Retryable = false});
+                        }
+                    }
+                }
+            } else {
+                SetAccessServiceBulkAuthorizeError(key, record, response->Status.Msg, IsRetryableGrpcError(response->Status));
+            }
+            Respond(record);
+        }
+    }
+
     void Handle(NCloud::TEvAccessService::TEvAuthorizeResponse::TPtr& ev) {
         NCloud::TEvAccessService::TEvAuthorizeResponse* response = ev->Get();
         TEvAccessServiceAuthorizeRequest* request = response->Request->Get<TEvAccessServiceAuthorizeRequest>();
@@ -755,12 +939,12 @@ private:
             auto& record = itToken->second;
             TString permission = request->Request.permission();
             TString subject;
-            yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase subjectType = yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase::TYPE_NOT_SET;
+            typename TPermissionRecord::TTypeCase subjectType = TPermissionRecord::TTypeCase::TYPE_NOT_SET;
             auto itPermission = record.Permissions.find(permission);
             if (itPermission != record.Permissions.end()) {
                 if (response->Status.Ok()) {
                     subject = GetSubjectName(response->Response.subject());
-                    subjectType = response->Response.subject().type_case();
+                    subjectType = ConvertSubjectType(response->Response.subject().type_case());
                     itPermission->second.Subject = subject;
                     itPermission->second.SubjectType = subjectType;
                     itPermission->second.Error.clear();
@@ -836,7 +1020,7 @@ private:
                 if (permissionsOk > 0 && retryableErrors == 0 && !requiredPermissionFailed) {
                     record.TokenType = request->Request.has_api_key() ? TDerived::ETokenType::ApiKey : TDerived::ETokenType::AccessService;
                     switch (subjectType) {
-                    case yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase::kUserAccount:
+                    case TPermissionRecord::TTypeCase::USER_ACCOUNT_TYPE:
                         if (UserAccountService) {
                             BLOG_TRACE("Ticket " << record.GetMaskedTicket()
                                         << " asking for UserAccount(" << subject << ")");
@@ -848,7 +1032,7 @@ private:
                             return;
                         }
                         break;
-                    case yandex::cloud::priv::servicecontrol::v1::Subject::TypeCase::kServiceAccount:
+                    case TPermissionRecord::TTypeCase::SERVICE_ACCOUNT_TYPE:
                         if (ServiceAccountService) {
                             BLOG_TRACE("Ticket " << record.GetMaskedTicket()
                                         << " asking for ServiceAccount(" << subject << ")");
@@ -1054,13 +1238,13 @@ protected:
         }
 
         if (tokenType == "Bearer" || tokenType == "IAM") {
-            if (AccessServiceValidator) {
+            if (AccessServiceValidatorV1 && AccessServiceValidatorV2) {
                 return TDerived::ETokenType::AccessService;
             } else {
                 return TDerived::ETokenType::Unsupported;
             }
         } else if (tokenType == "ApiKey") {
-            if (AccessServiceValidator && Config.GetUseAccessServiceApiKey()) {
+            if (AccessServiceValidatorV1 && AccessServiceValidatorV2 && Config.GetUseAccessServiceApiKey()) {
                 return TDerived::ETokenType::ApiKey;
             } else {
                 return TDerived::ETokenType::Unsupported;
@@ -1116,7 +1300,7 @@ protected:
     template <typename TTokenRecord>
     void InitTokenRecord(const TString& key, TTokenRecord& record, TInstant) {
         if (GetDerived()->CanInitAccessServiceToken(record)) {
-            if (AccessServiceValidator) {
+            if (AccessServiceValidatorV1 && AccessServiceValidatorV2) {
                 if (record.Permissions) {
                     RequestAccessServiceAuthorization(key, record);
                 } else {
@@ -1278,7 +1462,7 @@ protected:
 
     template <typename TTokenRecord>
     bool CanRefreshAccessServiceTicket(const TTokenRecord& record) {
-        if (!AccessServiceValidator) {
+        if (!AccessServiceValidatorV1 && AccessServiceValidatorV2) {
             return false;
         }
         if (record.TokenType == TDerived::ETokenType::AccessService || record.TokenType == TDerived::ETokenType::ApiKey) {
@@ -1397,7 +1581,7 @@ protected:
 
     void WriteAuthorizeMethods(TStringBuilder& html) {
         html << "<tr><td>Login</td><td>" << HtmlBool(UseLoginProvider) << "</td></tr>";
-        html << "<tr><td>Access Service</td><td>" << HtmlBool((bool)AccessServiceValidator) << "</td></tr>";
+        html << "<tr><td>Access Service</td><td>" << HtmlBool((bool)AccessServiceValidatorV1 && (bool)AccessServiceValidatorV2) << "</td></tr>";
         html << "<tr><td>User Account Service</td><td>" << HtmlBool((bool)UserAccountService) << "</td></tr>";
         html << "<tr><td>Service Account Service</td><td>" << HtmlBool((bool)ServiceAccountService) << "</td></tr>";
     }
@@ -1444,21 +1628,23 @@ protected:
             }
             settings.GrpcKeepAliveTimeMs = Config.GetAccessServiceGrpcKeepAliveTimeMs();
             settings.GrpcKeepAliveTimeoutMs = Config.GetAccessServiceGrpcKeepAliveTimeoutMs();
-            AccessServiceValidator = Register(NCloud::CreateAccessService(settings), TMailboxType::HTSwap, AppData()->UserPoolId);
+            AccessServiceValidatorV1 = Register(NCloud::CreateAccessServiceV1(settings), TMailboxType::HTSwap, AppData()->UserPoolId);
             if (Config.GetCacheAccessServiceAuthentication()) {
-                AccessServiceValidator = Register(NCloud::CreateGrpcServiceCache<NCloud::TEvAccessService::TEvAuthenticateRequest, NCloud::TEvAccessService::TEvAuthenticateResponse>(
-                                                          AccessServiceValidator,
+                AccessServiceValidatorV1 = Register(NCloud::CreateGrpcServiceCache<NCloud::TEvAccessService::TEvAuthenticateRequest, NCloud::TEvAccessService::TEvAuthenticateResponse>(
+                                                          AccessServiceValidatorV1,
                                                           Config.GetGrpcCacheSize(),
                                                           TDuration::MilliSeconds(Config.GetGrpcSuccessLifeTime()),
                                                           TDuration::MilliSeconds(Config.GetGrpcErrorLifeTime())), TMailboxType::HTSwap, AppData()->UserPoolId);
             }
             if (Config.GetCacheAccessServiceAuthorization()) {
-                AccessServiceValidator = Register(NCloud::CreateGrpcServiceCache<NCloud::TEvAccessService::TEvAuthorizeRequest, NCloud::TEvAccessService::TEvAuthorizeResponse>(
-                                                          AccessServiceValidator,
+                AccessServiceValidatorV1 = Register(NCloud::CreateGrpcServiceCache<NCloud::TEvAccessService::TEvAuthorizeRequest, NCloud::TEvAccessService::TEvAuthorizeResponse>(
+                                                          AccessServiceValidatorV1,
                                                           Config.GetGrpcCacheSize(),
                                                           TDuration::MilliSeconds(Config.GetGrpcSuccessLifeTime()),
                                                           TDuration::MilliSeconds(Config.GetGrpcErrorLifeTime())), TMailboxType::HTSwap, AppData()->UserPoolId);
             }
+
+            AccessServiceValidatorV2 = Register(NCloud::CreateAccessServiceV2(settings), TMailboxType::HTSwap, AppData()->UserPoolId);
         }
 
         if (Config.GetUseUserAccountService()) {
@@ -1509,8 +1695,11 @@ protected:
     }
 
     void PassAway() override {
-        if (AccessServiceValidator) {
-            Send(AccessServiceValidator, new TEvents::TEvPoisonPill);
+        if (AccessServiceValidatorV1) {
+            Send(AccessServiceValidatorV1, new TEvents::TEvPoisonPill);
+        }
+        if (AccessServiceValidatorV2) {
+            Send(AccessServiceValidatorV2, new TEvents::TEvPoisonPill);
         }
         if (UserAccountService) {
             Send(UserAccountService, new TEvents::TEvPoisonPill);
@@ -1556,6 +1745,7 @@ public:
             hFunc(TEvLdapAuthProvider::TEvEnrichGroupsResponse, Handle);
             hFunc(NCloud::TEvAccessService::TEvAuthenticateResponse, Handle);
             hFunc(NCloud::TEvAccessService::TEvAuthorizeResponse, Handle);
+            hFunc(NCloud::TEvAccessService::TEvBulkAuthorizeResponse, Handle);
             hFunc(NCloud::TEvUserAccountService::TEvGetUserAccountResponse, Handle);
             hFunc(NCloud::TEvServiceAccountService::TEvGetServiceAccountResponse, Handle);
             hFunc(NMon::TEvHttpInfo, Handle);
diff --git a/ydb/core/security/ticket_parser_ut.cpp b/ydb/core/security/ticket_parser_ut.cpp
index 9f3fa8b862f..57e87fd9e1b 100644
--- a/ydb/core/security/ticket_parser_ut.cpp
+++ b/ydb/core/security/ticket_parser_ut.cpp
@@ -1158,7 +1158,8 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT_VALUES_EQUAL(result->Token->GetUserSID(), "user1@as");
     }
 
-    Y_UNIT_TEST(AuthorizationRetryError) {
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void AuthorizationRetryError() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1174,6 +1175,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetUseStaff(false);
         authConfig.SetMinErrorRefreshTime("300ms");
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1185,7 +1187,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         client.InitRootScheme();
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder;
         builder.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder.BuildAndStart());
@@ -1220,7 +1222,16 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT(!result->Token->IsExist("something.write-bbbb4554@as"));
     }
 
-    Y_UNIT_TEST(AuthorizationRetryErrorImmediately) {
+    Y_UNIT_TEST(AuthorizationRetryError) {
+        AuthorizationRetryError<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorizationRetryError) {
+        AuthorizationRetryError<TTicketParserAccessServiceMockV2, true>();
+    }
+
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void AuthorizationRetryErrorImmediately() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1236,6 +1247,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetUseStaff(false);
         authConfig.SetRefreshPeriod("5s");
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1247,7 +1259,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         client.InitRootScheme();
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder;
         builder.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder.BuildAndStart());
@@ -1280,6 +1292,14 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT(!result->Token->IsExist("something.write-bbbb4554@as"));
     }
 
+    Y_UNIT_TEST(AuthorizationRetryErrorImmediately) {
+        AuthorizationRetryErrorImmediately<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorizationRetryErrorImmediately) {
+        AuthorizationRetryErrorImmediately<TTicketParserAccessServiceMockV2, true>();
+    }
+
     Y_UNIT_TEST(AuthenticationUnsupported) {
         using namespace Tests;
 
@@ -1371,7 +1391,8 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT_VALUES_EQUAL(result->Error.Message, "Unknown token");
     }
 
-    Y_UNIT_TEST(Authorization) {
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void Authorization() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1387,6 +1408,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetAccessServiceEndpoint(accessServiceEndpoint);
         authConfig.SetUseStaff(false);
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1400,7 +1422,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         TString userToken = "user1";
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder;
         builder.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder.BuildAndStart());
@@ -1419,6 +1441,18 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT(result->Token->IsExist("something.read-bbbb4554@as"));
         UNIT_ASSERT(!result->Token->IsExist("something.write-bbbb4554@as"));
 
+        accessServiceMock.AllowedUserPermissions.insert("user1-something.connect");
+        runtime->Send(new IEventHandle(MakeTicketParserID(), sender, new TEvTicketParser::TEvAuthorizeTicket(
+                                           userToken,
+                                           {{"folder_id", "aaaa1234"}, {"database_id", "bbbb4554"}},
+                                           {"something.read", "something.connect", "something.list", "something.update"})), 0);
+        result = runtime->GrabEdgeEvent<TEvTicketParser::TEvAuthorizeTicketResult>(handle);
+        UNIT_ASSERT(result->Error.empty());
+        UNIT_ASSERT(result->Token->IsExist("something.read-bbbb4554@as"));
+        UNIT_ASSERT(result->Token->IsExist("something.connect-bbbb4554@as"));
+        UNIT_ASSERT(!result->Token->IsExist("something.list-bbbb4554@as"));
+        UNIT_ASSERT(!result->Token->IsExist("something.update-bbbb4554@as"));
+
         // Authorization ApiKey successful.
         runtime->Send(new IEventHandle(MakeTicketParserID(), sender, new TEvTicketParser::TEvAuthorizeTicket(
                                            "ApiKey ApiKey-value-valid",
@@ -1514,7 +1548,16 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT(result->Token->IsExist("monitoring.view-gizmo@as"));
     }
 
-    Y_UNIT_TEST(AuthorizationWithRequiredPermissions) {
+    Y_UNIT_TEST(Authorization) {
+        Authorization<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorization) {
+        Authorization<TTicketParserAccessServiceMockV2, true>();
+    }
+
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void AuthorizationWithRequiredPermissions() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1529,6 +1572,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetAccessServiceEndpoint(accessServiceEndpoint);
         authConfig.SetUseStaff(false);
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1542,7 +1586,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         TString userToken = "user1";
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder;
         builder.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder.BuildAndStart());
@@ -1572,7 +1616,16 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT_VALUES_EQUAL(result->Error.Message, "something.write for folder_id aaaa1234 - Access Denied");
     }
 
-    Y_UNIT_TEST(AuthorizationWithUserAccount) {
+    Y_UNIT_TEST(AuthorizationWithRequiredPermissions) {
+        AuthorizationWithRequiredPermissions<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorizationWithRequiredPermissions) {
+        AuthorizationWithRequiredPermissions<TTicketParserAccessServiceMockV2, true>();
+    }
+
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void AuthorizationWithUserAccount() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1593,6 +1646,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetCacheAccessServiceAuthorization(false);
         //
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1606,7 +1660,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         TString userToken = "user1";
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder1;
         builder1.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder1.BuildAndStart());
@@ -1670,7 +1724,16 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT_VALUES_EQUAL(result->Token->GetUserSID(), "login1@passport");
     }
 
-    Y_UNIT_TEST(AuthorizationWithUserAccount2) {
+    Y_UNIT_TEST(AuthorizationWithUserAccount) {
+        AuthorizationWithUserAccount<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorizationWithUserAccount) {
+        AuthorizationWithUserAccount<TTicketParserAccessServiceMockV2, true>();
+    }
+
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void AuthorizationWithUserAccount2() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1688,6 +1751,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetUseUserAccountServiceTLS(false);
         authConfig.SetUserAccountServiceEndpoint(userAccountServiceEndpoint);
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1701,7 +1765,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         TString userToken = "user1";
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder1;
         builder1.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder1.BuildAndStart());
@@ -1735,7 +1799,16 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT_VALUES_EQUAL(result->Token->GetUserSID(), "login1@passport");
     }
 
-    Y_UNIT_TEST(AuthorizationUnavailable) {
+    Y_UNIT_TEST(AuthorizationWithUserAccount2) {
+        AuthorizationWithUserAccount2<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorizationWithUserAccount2) {
+        AuthorizationWithUserAccount2<TTicketParserAccessServiceMockV2, true>();
+    }
+
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void AuthorizationUnavailable() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1750,6 +1823,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetAccessServiceEndpoint(accessServiceEndpoint);
         authConfig.SetUseStaff(false);
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1763,7 +1837,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         TString userToken = "user1";
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder;
         builder.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder.BuildAndStart());
@@ -1785,7 +1859,16 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT_VALUES_EQUAL(result->Error.Message, "Service Unavailable");
     }
 
-    Y_UNIT_TEST(AuthorizationModify) {
+    Y_UNIT_TEST(AuthorizationUnavailable) {
+        AuthorizationUnavailable<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorizationUnavailable) {
+        AuthorizationUnavailable<TTicketParserAccessServiceMockV2, true>();
+    }
+
+    template <typename TAccessServiceMock, bool EnableBulkAuthorization = false>
+    void AuthorizationModify() {
         using namespace Tests;
 
         TPortManager tp;
@@ -1800,6 +1883,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         authConfig.SetAccessServiceEndpoint(accessServiceEndpoint);
         authConfig.SetUseStaff(false);
         auto settings = TServerSettings(port, authConfig);
+        settings.SetEnableAccessServiceBulkAuthorization(EnableBulkAuthorization);
         settings.SetDomainName("Root");
         settings.CreateTicketParser = NKikimr::CreateTicketParser;
         TServer server(settings);
@@ -1813,7 +1897,7 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         TString userToken = "user1";
 
         // Access Server Mock
-        NKikimr::TAccessServiceMock accessServiceMock;
+        TAccessServiceMock accessServiceMock;
         grpc::ServerBuilder builder;
         builder.AddListeningPort(accessServiceEndpoint, grpc::InsecureServerCredentials()).RegisterService(&accessServiceMock);
         std::unique_ptr<grpc::Server> accessServer(builder.BuildAndStart());
@@ -1845,5 +1929,13 @@ Y_UNIT_TEST_SUITE(TTicketParserTest) {
         UNIT_ASSERT(result->Token->IsExist("something.read-bbbb4554@as"));
         UNIT_ASSERT(result->Token->IsExist("something.write-bbbb4554@as"));
     }
+
+    Y_UNIT_TEST(AuthorizationModify) {
+        AuthorizationModify<NKikimr::TAccessServiceMock>();
+    }
+
+    Y_UNIT_TEST(BulkAuthorizationModify) {
+        AuthorizationModify<TTicketParserAccessServiceMockV2, true>();
+    }
 }
 }
diff --git a/ydb/core/testlib/basics/feature_flags.h b/ydb/core/testlib/basics/feature_flags.h
index 94ebd910a20..3185f6ef877 100644
--- a/ydb/core/testlib/basics/feature_flags.h
+++ b/ydb/core/testlib/basics/feature_flags.h
@@ -55,6 +55,7 @@ public:
     FEATURE_FLAG_SETTER(EnableUuidAsPrimaryKey)
     FEATURE_FLAG_SETTER(EnableTablePgTypes)
     FEATURE_FLAG_SETTER(EnableServerlessExclusiveDynamicNodes)
+    FEATURE_FLAG_SETTER(EnableAccessServiceBulkAuthorization)
 
     #undef FEATURE_FLAG_SETTER
 };
diff --git a/ydb/library/testlib/service_mocks/access_service_mock.h b/ydb/library/testlib/service_mocks/access_service_mock.h
index 927f7fdc480..5d6f421472e 100644
--- a/ydb/library/testlib/service_mocks/access_service_mock.h
+++ b/ydb/library/testlib/service_mocks/access_service_mock.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ydb/public/api/client/yc_private/servicecontrol/access_service.grpc.pb.h>
+#include <ydb/public/api/client/yc_private/accessservice/access_service.grpc.pb.h>
 
 #include <library/cpp/testing/unittest/registar.h>
 
@@ -186,3 +187,99 @@ public:
         }
     }
 };
+
+class TTicketParserAccessServiceMockV2 : public yandex::cloud::priv::accessservice::v2::AccessService::Service {
+public:
+    std::atomic_uint64_t AuthorizeCount= 0;
+    bool ShouldGenerateRetryableError = false;
+    bool ShouldGenerateOneRetryableError = false;
+    bool isUserAuthenticated = true;
+
+    THashSet<TString> UnavailableUserPermissions;
+    THashSet<TString> AllowedResourceIds;
+    THashSet<TString> AllowedUserPermissions = {
+        "user1-something.read",
+        "ApiKey-value-valid-something.read",
+        "ApiKey-value-valid-ydb.api.kafkaPlainAuth",
+        "user1-monitoring.view"
+    };
+    THashMap<TString, TString> AllowedServicePermissions = {{"service1-something.write", "root1/folder1"}};
+
+public:
+    ::grpc::Status BulkAuthorize(::grpc::ServerContext*,
+                                 const ::yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest* request,
+                                 ::yandex::cloud::priv::accessservice::v2::BulkAuthorizeResponse* response) {
+        ++AuthorizeCount;
+        if (request->has_signature()) {
+            if (ShouldGenerateRetryableError) {
+                return grpc::Status(grpc::StatusCode::UNAVAILABLE, "Service Unavailable");
+            }
+            if (ShouldGenerateOneRetryableError) {
+                ShouldGenerateOneRetryableError = false;
+                return grpc::Status(grpc::StatusCode::UNAVAILABLE, "Service Unavailable");
+            }
+            response->mutable_subject()->mutable_user_account()->set_id("user1");
+            return grpc::Status::OK;
+        } else {
+            if (!isUserAuthenticated) {
+                auto error = response->mutable_unauthenticated_error();
+                error->set_message("Access Denied");
+                return grpc::Status(grpc::StatusCode::UNAUTHENTICATED, "Access Denied");
+            }
+            TString token = request->has_iam_token() ? request->iam_token() : request->api_key();
+            if (request->has_actions()) {
+                const auto& actions = request->actions();
+                bool wasFoundFirstAccessDenied = false;
+                for (const auto& action : actions.items()) {
+                    if (UnavailableUserPermissions.count(token + '-' + action.permission()) > 0) {
+                        return grpc::Status(grpc::StatusCode::UNAVAILABLE, "Service Unavailable");
+                    }
+
+                    bool allowedResource = true;
+                    if (!AllowedResourceIds.empty()) {
+                        allowedResource = false;
+                        for (const auto& resourcePath : action.resource_path()) {
+                            if (AllowedResourceIds.count(resourcePath.id()) > 0) {
+                                allowedResource = true;
+                            }
+                        }
+                    }
+                    if (allowedResource) {
+                        if (AllowedUserPermissions.count(token + '-' + action.permission()) > 0) {
+                            response->mutable_subject()->mutable_user_account()->set_id(token);
+
+                        } else if (AllowedServicePermissions.count(token + '-' + action.permission()) > 0) {
+                            response->mutable_subject()->mutable_service_account()->set_id(token);
+                            response->mutable_subject()->mutable_service_account()->set_folder_id(AllowedServicePermissions[token + '-' + action.permission()]);
+                        } else {
+                            if (request->result_filter() == yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest::ALL_FAILED) {
+                                SetAccessDenied(response->mutable_results(), action);
+                            } else {
+                                if (!wasFoundFirstAccessDenied) {
+                                    SetAccessDenied(response->mutable_results(), action);
+                                    wasFoundFirstAccessDenied = true;
+                                }
+                            }
+                        }
+                    } else {
+                        SetAccessDenied(response->mutable_results(), action);
+                    }
+                }
+            }
+            return grpc::Status(grpc::StatusCode::OK, "OK");
+        }
+    }
+
+private:
+    void SetAccessDenied(::yandex::cloud::priv::accessservice::v2::BulkAuthorizeResponse_Results* results,
+                         const ::yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest_Action& action) {
+        auto result = results->add_items();
+        result->set_permission(action.permission());
+        for (const auto& resource_path : action.resource_path()) {
+            auto rp = result->add_resource_path();
+            rp->CopyFrom(resource_path);
+        }
+        auto error = result->mutable_permission_denied_error();
+        error->set_message("Access Denied");
+    }
+};
diff --git a/ydb/library/testlib/service_mocks/ya.make b/ydb/library/testlib/service_mocks/ya.make
index 9bf784d56f6..a8879802333 100644
--- a/ydb/library/testlib/service_mocks/ya.make
+++ b/ydb/library/testlib/service_mocks/ya.make
@@ -13,6 +13,7 @@ SRCS(
 
 PEERDIR(
     ydb/public/api/client/yc_private/servicecontrol
+    ydb/public/api/client/yc_private/accessservice
     ydb/public/api/grpc/draft
     ydb/public/api/client/yc_private/resourcemanager
     ydb/public/api/client/yc_private/iam
diff --git a/ydb/library/ycloud/api/access_service.h b/ydb/library/ycloud/api/access_service.h
index 67149a8e107..2c60a82d9e6 100644
--- a/ydb/library/ycloud/api/access_service.h
+++ b/ydb/library/ycloud/api/access_service.h
@@ -2,6 +2,7 @@
 #include <ydb/core/base/defs.h>
 #include <ydb/core/base/events.h>
 #include <ydb/public/api/client/yc_private/servicecontrol/access_service.grpc.pb.h>
+#include <ydb/public/api/client/yc_private/accessservice/access_service.grpc.pb.h>
 #include "events.h"
 
 namespace NCloud {
@@ -12,10 +13,12 @@ namespace NCloud {
             // requests
             EvAuthenticateRequest = EventSpaceBegin(TKikimrEvents::ES_ACCESS_SERVICE),
             EvAuthorizeRequest,
+            EvBulkAuthorizeRequest,
 
             // replies
             EvAuthenticateResponse = EventSpaceBegin(TKikimrEvents::ES_ACCESS_SERVICE) + 512,
             EvAuthorizeResponse,
+            EvBulkAuthorizeResponse,
 
             EvEnd
         };
@@ -29,5 +32,8 @@ namespace NCloud {
 
         struct TEvAuthorizeRequest : TEvGrpcProtoRequest<TEvAuthorizeRequest, EvAuthorizeRequest, yandex::cloud::priv::servicecontrol::v1::AuthorizeRequest> {};
         struct TEvAuthorizeResponse : TEvGrpcProtoResponse<TEvAuthorizeResponse, EvAuthorizeResponse, yandex::cloud::priv::servicecontrol::v1::AuthorizeResponse> {};
+
+        struct TEvBulkAuthorizeRequest : TEvGrpcProtoRequest<TEvBulkAuthorizeRequest, EvBulkAuthorizeRequest, yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest> {};
+        struct TEvBulkAuthorizeResponse : TEvGrpcProtoResponse<TEvBulkAuthorizeResponse, EvBulkAuthorizeResponse, yandex::cloud::priv::accessservice::v2::BulkAuthorizeResponse> {};
    };
 }
diff --git a/ydb/library/ycloud/api/ya.make b/ydb/library/ycloud/api/ya.make
index 9348f2dfb37..bec43952758 100644
--- a/ydb/library/ycloud/api/ya.make
+++ b/ydb/library/ycloud/api/ya.make
@@ -11,6 +11,7 @@ SRCS(
 PEERDIR(
     ydb/public/api/client/yc_private/iam
     ydb/public/api/client/yc_private/servicecontrol
+    ydb/public/api/client/yc_private/accessservice
     ydb/public/api/client/yc_private/resourcemanager
     ydb/library/actors/core
     ydb/library/grpc/client
diff --git a/ydb/library/ycloud/impl/access_service.cpp b/ydb/library/ycloud/impl/access_service.cpp
index 1678f459afc..97ab06e485f 100644
--- a/ydb/library/ycloud/impl/access_service.cpp
+++ b/ydb/library/ycloud/impl/access_service.cpp
@@ -2,6 +2,7 @@
 #include <ydb/library/actors/core/actor.h>
 #include <library/cpp/json/json_value.h>
 #include <ydb/public/api/client/yc_private/servicecontrol/access_service.grpc.pb.h>
+#include <ydb/public/api/client/yc_private/accessservice/access_service.grpc.pb.h>
 #include "access_service.h"
 #include "grpc_service_client.h"
 #include "grpc_service_cache.h"
@@ -10,9 +11,9 @@ namespace NCloud {
 
 using namespace NKikimr;
 
-class TAccessService : public NActors::TActor<TAccessService>, TGrpcServiceClient<yandex::cloud::priv::servicecontrol::v1::AccessService> {
-    using TThis = TAccessService;
-    using TBase = NActors::TActor<TAccessService>;
+class TAccessServiceV1 : public NActors::TActor<TAccessServiceV1>, TGrpcServiceClient<yandex::cloud::priv::servicecontrol::v1::AccessService> {
+    using TThis = TAccessServiceV1;
+    using TBase = NActors::TActor<TAccessServiceV1>;
 
     struct TAuthenticateRequest : TGrpcRequest {
         static constexpr auto Request = &yandex::cloud::priv::servicecontrol::v1::AccessService::Stub::AsyncAuthenticate;
@@ -68,7 +69,7 @@ class TAccessService : public NActors::TActor<TAccessService>, TGrpcServiceClien
 public:
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() { return NKikimrServices::TActivity::ACCESS_SERVICE_ACTOR; }
 
-    TAccessService(const TAccessServiceSettings& settings)
+    TAccessServiceV1(const TAccessServiceSettings& settings)
         : TBase(&TThis::StateWork)
         , TGrpcServiceClient(settings)
     {}
@@ -82,13 +83,62 @@ public:
     }
 };
 
+class TAccessServiceV2 : public NActors::TActor<TAccessServiceV2>, TGrpcServiceClient<yandex::cloud::priv::accessservice::v2::AccessService> {
+    using TThis = TAccessServiceV2;
+    using TBase = NActors::TActor<TAccessServiceV2>;
 
-IActor* CreateAccessService(const TAccessServiceSettings& settings) {
-    return new TAccessService(settings);
+    struct TBulkAuthorizeRequest : TGrpcRequest {
+        static constexpr auto Request = &yandex::cloud::priv::accessservice::v2::AccessService::Stub::AsyncBulkAuthorize;
+        using TRequestEventType = TEvAccessService::TEvBulkAuthorizeRequest;
+        using TResponseEventType = TEvAccessService::TEvBulkAuthorizeResponse;
+
+        static yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest Obfuscate(const yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest& p) {
+            yandex::cloud::priv::accessservice::v2::BulkAuthorizeRequest r(p);
+            if (r.iam_token()) {
+                r.set_iam_token(MaskToken(r.iam_token()));
+            }
+            if (r.api_key()) {
+                r.set_api_key(MaskToken(r.api_key()));
+            }
+            return r;
+        }
+
+        static const yandex::cloud::priv::accessservice::v2::BulkAuthorizeResponse& Obfuscate(const yandex::cloud::priv::accessservice::v2::BulkAuthorizeResponse& p) {
+            return p;
+        }
+    };
+
+    void Handle(TEvAccessService::TEvBulkAuthorizeRequest::TPtr& ev) {
+        MakeCall<TBulkAuthorizeRequest>(std::move(ev));
+    }
+
+public:
+    static constexpr NKikimrServices::TActivity::EType ActorActivityType() { return NKikimrServices::TActivity::ACCESS_SERVICE_ACTOR; }
+
+    TAccessServiceV2(const TAccessServiceSettings& settings)
+        : TBase(&TThis::StateWork)
+        , TGrpcServiceClient(settings)
+    {}
+
+    void StateWork(TAutoPtr<NActors::IEventHandle>& ev) {
+        switch (ev->GetTypeRewrite()) {
+            hFunc(TEvAccessService::TEvBulkAuthorizeRequest, Handle);
+            cFunc(TEvents::TSystem::PoisonPill, PassAway);
+        }
+    }
+};
+
+
+IActor* CreateAccessServiceV1(const TAccessServiceSettings& settings) {
+    return new TAccessServiceV1(settings);
+}
+
+IActor* CreateAccessServiceV2(const TAccessServiceSettings& settings) {
+    return new TAccessServiceV2(settings);
 }
 
 IActor* CreateAccessServiceWithCache(const TAccessServiceSettings& settings) {
-    IActor* accessService = CreateAccessService(settings);
+    IActor* accessService = CreateAccessServiceV1(settings);
     accessService = CreateGrpcServiceCache<TEvAccessService::TEvAuthenticateRequest, TEvAccessService::TEvAuthenticateResponse>(accessService);
     accessService = CreateGrpcServiceCache<TEvAccessService::TEvAuthorizeRequest, TEvAccessService::TEvAuthorizeResponse>(accessService);
     return accessService;
diff --git a/ydb/library/ycloud/impl/access_service.h b/ydb/library/ycloud/impl/access_service.h
index bdcf2b1c003..ee1cb03c8d7 100644
--- a/ydb/library/ycloud/impl/access_service.h
+++ b/ydb/library/ycloud/impl/access_service.h
@@ -8,12 +8,19 @@ using namespace NKikimr;
 
 struct TAccessServiceSettings : TGrpcClientSettings {};
 
-IActor* CreateAccessService(const TAccessServiceSettings& settings);
+IActor* CreateAccessServiceV1(const TAccessServiceSettings& settings);
+IActor* CreateAccessServiceV2(const TAccessServiceSettings& settings);
 
-inline IActor* CreateAccessService(const TString& endpoint) {
+inline IActor* CreateAccessServiceV1(const TString& endpoint) {
     TAccessServiceSettings settings;
     settings.Endpoint = endpoint;
-    return CreateAccessService(settings);
+    return CreateAccessServiceV1(settings);
+}
+
+inline IActor* CreateAccessServiceV2(const TString& endpoint) {
+    TAccessServiceSettings settings;
+    settings.Endpoint = endpoint;
+    return CreateAccessServiceV2(settings);
 }
 
 IActor* CreateAccessServiceWithCache(const TAccessServiceSettings& settings); // for compatibility with older code
diff --git a/ydb/library/ycloud/impl/mock_access_service.cpp b/ydb/library/ycloud/impl/mock_access_service.cpp
index 1e2f44cc204..1f518213028 100644
--- a/ydb/library/ycloud/impl/mock_access_service.cpp
+++ b/ydb/library/ycloud/impl/mock_access_service.cpp
@@ -1,7 +1,6 @@
 #include <ydb/library/actors/core/actorsystem.h>
 #include <ydb/library/actors/core/actor.h>
 #include <library/cpp/json/json_value.h>
-#include <ydb/public/api/client/yc_private/servicecontrol/access_service.grpc.pb.h>
 #include "access_service.h"
 #include "grpc_service_client.h"
 #include "grpc_service_cache.h"
diff --git a/ydb/public/api/client/yc_private/accessservice/access_service.proto b/ydb/public/api/client/yc_private/accessservice/access_service.proto
new file mode 100644
index 00000000000..83d7fa9e0b8
--- /dev/null
+++ b/ydb/public/api/client/yc_private/accessservice/access_service.proto
@@ -0,0 +1,259 @@
+syntax = "proto3";
+
+package yandex.cloud.priv.accessservice.v2;
+
+import "google/protobuf/timestamp.proto";
+import "google/protobuf/any.proto";
+import "google/protobuf/field_mask.proto";
+import "ydb/public/api/client/yc_private/accessservice/resource.proto";
+import "ydb/public/api/client/yc_private/accessservice/sensitive.proto";
+
+option go_package = "accessservice_v2";
+option java_outer_classname = "PAS";
+
+service AccessService {
+  // Verify the identity of a subject.
+  //
+  // gRPC error codes
+  //
+  // Ok:              the provided credentials are valid
+  // Unauthenticated: the provided credentials are invalid or may have expired
+  // InvalidArgument: the client specified an invalid argument (please note that this applies to the request in itself,
+  //                  not to the content of the request, i.e. you will get the InvalidArgument error if the message
+  //                  size exceeds the server limit but Unauthenticated if the token format is not recognized)
+  // Unavailable:     the service is currently unavailable, the client should retry again
+  // Internal:        the service is broken
+  //
+  // Please note that these do not include client-side errors (e.g. Cancelled, DeadlineExceeded, etc.)
+  rpc Authenticate (AuthenticateRequest) returns (AuthenticateResponse);
+
+  // Check if a subject is allowed to perform an action. This also authenticates the subject if any credentials are
+  // passed as an identity.
+  //
+  // gRPC error codes
+  //
+  // Ok:               the provided credentials (if any) are valid and the subject has permissions to access the
+  //                   specified resource
+  // Unauthenticated:  the provided credentials are invalid or may have expired
+  // PermissionDenied: the subject does not have permissions to access the specified resource
+  // InvalidArgument:  the client specified an invalid argument (please note that this applies to the request in itself,
+  //                   not to the content of the request, i.e. you will get the InvalidArgument error if the message
+  //                   size exceeds the server limit or the specified permission does not exist but Unauthenticated if
+  //                   the token format is not recognized)
+  // Unavailable:      the service is currently unavailable, the client should retry again
+  // Internal:         the service is broken
+  //
+  // Please note that these do not include client-side errors (e.g. Cancelled, DeadlineExceeded, etc.)
+  rpc Authorize (AuthorizeRequest) returns (AuthorizeResponse);
+
+  // Similar to Authorize, but requests multiple actions for one subject.
+  //
+  // gRPC error codes will be the same, except for these cases:
+  // - An Unauthenticated error of BulkAuthorizeRequest.identity is returned in
+  //   BulkAuthorizeResponse.unauthenticated_error.
+  // - All PermissionDenied of BulkAuthorizeRequest.authorizations are returned in
+  //   BulkAuthorizeResponse.results.
+  //
+  // You can control the information returned in BulkAuthorizeResponse.results with:
+  // - result_filter : return all errors (ALL_FAILED) or only the first one (FIRST_FAILED), if any.
+  // - result_mask : You can choose the fields returned (all by default),
+  //     from the fields in BulkAuthorizeResponse.Result.
+  //
+  rpc BulkAuthorize (BulkAuthorizeRequest) returns (BulkAuthorizeResponse);
+}
+
+message AuthenticateRequest {
+  oneof credentials {
+    // option (exactly_one) = true;
+
+    // IAM-token obtained from the IAM Token Service.
+    // The server response for an empty IAM token is UNAUTHENTICATED
+    string iam_token = 1 [(sensitive) = true, (sensitive_type) = SENSITIVE_IAM_TOKEN]; // [(length) = "<=1024"];
+
+    // AWS-compatible signature.
+    AccessKeySignature signature = 2;
+
+    // API key.
+    // The server response for an empty API key is UNAUTHENTICATED
+    string api_key = 3 [(sensitive) = true, (sensitive_type) = SENSITIVE_CRC];
+
+    // IAM-cookie.
+    // The server response for an empty IAM cookie is UNAUTHENTICATED
+    string iam_cookie = 4 [(sensitive) = true, (sensitive_type) = SENSITIVE_IAM_COOKIE];
+
+    // RefreshToken.
+    // The server response for an empty RefreshToken is UNAUTHENTICATED
+    string refresh_token = 5 [(sensitive) = true, (sensitive_type) = SENSITIVE_REFRESH_TOKEN];
+  }
+}
+
+message AuthenticateResponse {
+  Subject subject = 1;
+}
+
+message AuthorizeRequest {
+  oneof identity {
+    // option (exactly_one) = true;
+
+    Subject subject = 1;
+
+    // IAM-token obtained from the IAM Token Service.
+    // The server response for an empty IAM token is UNAUTHENTICATED
+    string iam_token = 2 [(sensitive) = true, (sensitive_type) = SENSITIVE_IAM_TOKEN]; // [(length) = "<=1024"];
+
+    // AWS-compatible signature.
+    AccessKeySignature signature = 3;
+
+    // API key.
+    // The server response for an empty API key is UNAUTHENTICATED
+    string api_key = 6 [(sensitive) = true, (sensitive_type) = SENSITIVE_CRC];
+  }
+
+  string permission = 4; // [(required) = true, (length) = "<=128"];
+
+  // A resource to authorize access to. This may also include a service-specific hierarchy of the resource, usually
+  // ends with resource-manager.folder.
+  //
+  // Examples:
+  //     (resource-manager.folder, b1gn3enigctah04o0fkb)
+  //     (billing.account, b1gqql62454n46tboesn)
+  //     (compute.instance, b1gqqhvc4fg65mkrefs8), (resource-manager.folder, b1gn3enigctah04o0fkb)
+  //     (resource-manager.cloud, aje56o8prppkrpaiuoc6)
+  //     (my-service.instance, b1gqqepv0upu57issrog), (resource-manager.cloud, aje56o8prppkrpaiuoc6)
+  repeated Resource resource_path = 5; // [(size) = ">0"];
+}
+
+message AuthorizeResponse {
+  Subject subject = 1;
+
+  // Full path to the resource.
+  repeated Resource resource_path = 2;
+}
+
+message BulkAuthorizeRequest {
+  oneof identity {
+    // option (exactly_one) = true;
+
+    Subject subject = 1;
+
+    string iam_token = 2 [(sensitive) = true, (sensitive_type) = SENSITIVE_IAM_TOKEN]; // [(length) = "<=1024"];
+
+    AccessKeySignature signature = 3;
+
+    string api_key = 4 [(sensitive) = true, (sensitive_type) = SENSITIVE_CRC];
+  }
+
+  oneof authorizations {
+    // option (exactly_one) = true;
+
+    Actions actions = 5;
+
+    ActionMatrix action_matrix = 6;
+  }
+
+  ResultFilter result_filter = 7;
+
+  google.protobuf.FieldMask result_mask = 8;
+
+  message Action {
+    repeated Resource resource_path = 1; // [(size) = "1-128"];
+
+    string permission = 2; // [(required) = true, (length) = "<=128"];
+  }
+
+  message Actions {
+    repeated Action items = 1; // [(size) = "1-1000"];
+  }
+
+  // Cross product of paths and permissions (represents N*M actions, N*M <= 1000).
+  message ActionMatrix {
+    repeated ResourcePath resource_paths = 2; // [(size) = "1-1000"];
+
+    repeated string permissions = 1; // [(size) = "1-1000", (length) = "<=128"];
+  }
+
+  enum ResultFilter {
+    RESULT_FILTER_UNSPECIFIED = 0;
+    FIRST_FAILED = 1;
+    ALL_FAILED = 2;
+  }
+}
+
+message BulkAuthorizeResponse {
+  Subject subject = 1;
+
+  Error unauthenticated_error = 2;
+
+  Results results = 3;
+
+  message Results {
+    repeated Result items = 1;
+  }
+
+  message Result {
+    string permission = 1;
+
+    repeated Resource resource_path = 2;
+
+    Error permission_denied_error = 3;
+  }
+
+  message Error {
+    string message = 1;
+
+    repeated google.protobuf.Any details = 2;
+  }
+}
+
+message AccessKeySignature {
+  string access_key_id = 1; // [(required) = true, (length) = "<=50"];
+  string string_to_sign = 2; // [(required) = true, (length) = "<=8192"];
+  string signature = 3 [(sensitive) = true, (sensitive_type) = SENSITIVE_CRC]; // [(required) = true, (length) = "<=128"];
+
+  oneof parameters {
+    // option (exactly_one) = true;
+
+    Version2Parameters v2_parameters = 4;
+    Version4Parameters v4_parameters = 5;
+  }
+
+  message Version2Parameters {
+    SignatureMethod signature_method = 1;
+
+    enum SignatureMethod {
+      SIGNATURE_METHOD_UNSPECIFIED = 0;
+      HMAC_SHA1 = 1;
+      HMAC_SHA256 = 2;
+    }
+  }
+
+  message Version4Parameters {
+    google.protobuf.Timestamp signed_at = 1; // [(required) = true];
+    string service = 2; // [(required) = true, (length) = "<=64"];
+    string region = 3; // [(required) = true, (length) = "<=32"];
+  }
+}
+
+message Subject {
+  oneof type {
+    // option (exactly_one) = true;
+
+    UserAccount user_account = 1;
+    ServiceAccount service_account = 2;
+    AnonymousAccount anonymous_account = 3;
+  }
+
+  message UserAccount {
+    string id = 1; // [(required) = true, (length) = "<=50"];
+    string federation_id = 2; // [(length) = "<=50"];
+  }
+
+  message ServiceAccount {
+    string id = 1; // [(required) = true, (length) = "<=50"];
+    string folder_id = 2; // [(length) = "<=50"];
+  }
+
+  // Use this if you want to check if an unauthenticated subject is allowed to access a resource.
+  message AnonymousAccount {
+  }
+}
diff --git a/ydb/public/api/client/yc_private/accessservice/resource.proto b/ydb/public/api/client/yc_private/accessservice/resource.proto
new file mode 100644
index 00000000000..a20fb36cf1e
--- /dev/null
+++ b/ydb/public/api/client/yc_private/accessservice/resource.proto
@@ -0,0 +1,17 @@
+syntax = "proto3";
+
+package yandex.cloud.priv.accessservice.v2;
+
+option go_package = "accessservice_v2";
+option java_outer_classname = "PR";
+
+message Resource {
+  string id = 1; // [(required) = true, (length) = "<=50"];
+
+  // The type of the resource, e.g. resource-manager.folder, billing.account, compute.snapshot, etc.
+  string type = 2; // [(required) = true, (length) = "<=64"];
+}
+
+message ResourcePath {
+  repeated Resource path = 1; // [(size) = "1-128"];
+}
diff --git a/ydb/public/api/client/yc_private/accessservice/sensitive.proto b/ydb/public/api/client/yc_private/accessservice/sensitive.proto
new file mode 100644
index 00000000000..009620bd991
--- /dev/null
+++ b/ydb/public/api/client/yc_private/accessservice/sensitive.proto
@@ -0,0 +1,29 @@
+syntax = "proto3";
+
+// Based on:
+// https://bb.yandexcloud.net/projects/CLOUD/repos/cloud-go/browse/private-api/yandex/cloud/priv/sensitive.proto
+
+package yandex.cloud;
+
+import "google/protobuf/descriptor.proto";
+
+option go_package = "cloud/proto_extensions";
+
+enum SensitiveType {
+  SENSITIVE_TYPE_UNSPECIFIED = 0;
+  SENSITIVE_CRC = 1;
+  SENSITIVE_IAM_TOKEN = 2;
+  SENSITIVE_REMOVE = 3;
+  SENSITIVE_YANDEX_PASSPORT_OAUTH_TOKEN = 4;
+  SENSITIVE_IAM_COOKIE = 5;
+  SENSITIVE_REFRESH_TOKEN = 6;
+  SENSITIVE_SESSION_TOKEN = 7;
+}
+
+extend google.protobuf.FieldOptions {
+  // novikoff:
+  // Sensitive fields are hidden in logs
+  // For now could be applied only to string fields
+  bool sensitive = 110601;
+  SensitiveType sensitive_type = 110602;
+}
diff --git a/ydb/public/api/client/yc_private/accessservice/ya.make b/ydb/public/api/client/yc_private/accessservice/ya.make
new file mode 100644
index 00000000000..8199005d233
--- /dev/null
+++ b/ydb/public/api/client/yc_private/accessservice/ya.make
@@ -0,0 +1,16 @@
+PROTO_LIBRARY()
+
+EXCLUDE_TAGS(GO_PROTO)
+
+GRPC()
+SRCS(
+    access_service.proto
+    resource.proto
+    sensitive.proto
+)
+
+USE_COMMON_GOOGLE_APIS(
+    api/annotations
+)
+
+END()
diff --git a/ydb/public/api/client/yc_private/servicecontrol/ya.make b/ydb/public/api/client/yc_private/servicecontrol/ya.make
index 1820f9a20b9..15a84d453dd 100644
--- a/ydb/public/api/client/yc_private/servicecontrol/ya.make
+++ b/ydb/public/api/client/yc_private/servicecontrol/ya.make
@@ -13,4 +13,3 @@ USE_COMMON_GOOGLE_APIS(
 )
 
 END()
-