Skip to content

Commit 69a62d6

Browse files
alexv-smirnovalexv-smirnov
authored
stream-nb-24-4: Merge audit tests with canonization (#26033)
2 parents c033996 + 87c7fd5 commit 69a62d6

File tree

37 files changed

+1007
-149
lines changed

37 files changed

+1007
-149
lines changed

‎ydb/apps/dstool/lib/common.py‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -117,10 +117,12 @@ def parse_token(self, token_file):
117117
except Exception:
118118
pass
119119

120-
if self.token is not None and len(self.token.split(' ')) == 2:
121-
self.token_type, self.token = self.token.split(' ')
122-
else:
123-
self.token_type = 'OAuth'
120+
if self.token is not None:
121+
if len(self.token.split(' ')) == 2:
122+
self.token_type, self.token = self.token.split(' ')
123+
else:
124+
if not self.token.endswith('@builtin'):
125+
self.token_type = 'OAuth'
124126

125127
def apply_args(self, args, with_localhost=True):
126128
self.grpc_port = args.grpc_port
@@ -303,7 +305,11 @@ def fetch(path, params={}, explicit_host=None, fmt='json', host=None, cache=True
303305
print('INFO: fetching %s' % url, file=sys.stderr)
304306
request = urllib.request.Request(url, data=data, method=method)
305307
if connection_params.token and url.startswith('http'):
306-
request.add_header('Authorization', '%s %s' % (connection_params.token_type, connection_params.token))
308+
if connection_params.token_type:
309+
authorization = '%s %s' % (connection_params.token_type, connection_params.token)
310+
else:
311+
authorization = connection_params.token
312+
request.add_header('Authorization', authorization)
307313
if content_type is not None:
308314
request.add_header('Content-Type', content_type)
309315
if accept is not None:

‎ydb/core/cms/console/console_configs_manager.cpp‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -976,21 +976,23 @@ void TConfigsManager::ScheduleLogCleanup(const TActorContext &ctx)
976976
}
977977

978978
void TConfigsManager::HandleUnauthorized(TEvConsole::TEvReplaceYamlConfigRequest::TPtr &ev, const TActorContext &) {
979+
NACLib::TUserToken token(ev->Get()->Record.GetUserToken());
979980
AuditLogReplaceConfigTransaction(
980981
/* peer = */ ev->Get()->Record.GetPeerName(),
981-
/* userSID = */ ev->Get()->Record.GetUserToken(),
982-
/* sanitizedToken = */ TString(),
982+
/* userSID = */ token.GetUserSID(),
983+
/* sanitizedToken = */ token.GetSanitizedToken(),
983984
/* oldConfig = */ YamlConfig,
984985
/* newConfig = */ ev->Get()->Record.GetRequest().config(),
985986
/* reason = */ "Unauthorized.",
986987
/* success = */ false);
987988
}
988989

989990
void TConfigsManager::HandleUnauthorized(TEvConsole::TEvSetYamlConfigRequest::TPtr &ev, const TActorContext &) {
991+
NACLib::TUserToken token(ev->Get()->Record.GetUserToken());
990992
AuditLogReplaceConfigTransaction(
991993
/* peer = */ ev->Get()->Record.GetPeerName(),
992-
/* userSID = */ ev->Get()->Record.GetUserToken(),
993-
/* sanitizedToken = */ TString(),
994+
/* userSID = */ token.GetUserSID(),
995+
/* sanitizedToken = */ token.GetSanitizedToken(),
994996
/* oldConfig = */ YamlConfig,
995997
/* newConfig = */ ev->Get()->Record.GetRequest().config(),
996998
/* reason = */ "Unauthorized.",

‎ydb/core/cms/console/console_tenants_manager.cpp‎

Lines changed: 17 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -536,16 +536,23 @@ class TSubDomainManip : public TActorBootstrapped<TSubDomainManip> {
536536
}
537537
}
538538

539-
void AlterUserAttribute(const TActorContext &ctx) {
540-
BLOG_D("TSubDomainManip(" << Tenant->Path << ") alter user attribute ");
539+
THolder<TEvTxUserProxy::TEvProposeTransaction> MakeProposeTransaction() {
541540
auto request = MakeHolder<TEvTxUserProxy::TEvProposeTransaction>();
542541

543542
request->Record.SetDatabaseName(TString(ExtractDomain(Subdomain.first)));
544543
request->Record.SetExecTimeoutPeriod(Max<ui64>());
544+
//request->Record.SetPeerName(Tenant->PeerName);
545545

546546
if (Tenant->UserToken.GetUserSID())
547547
request->Record.SetUserToken(Tenant->UserToken.SerializeAsString());
548548

549+
return request;
550+
}
551+
552+
void AlterUserAttribute(const TActorContext &ctx) {
553+
BLOG_D("TSubDomainManip(" << Tenant->Path << ") alter user attribute ");
554+
auto request = MakeProposeTransaction();
555+
549556
auto &tx = *request->Record.MutableTransaction()->MutableModifyScheme();
550557
tx.SetWorkingDir(Subdomain.first);
551558

@@ -563,13 +570,7 @@ class TSubDomainManip : public TActorBootstrapped<TSubDomainManip> {
563570
void AlterSubdomain(const TActorContext &ctx)
564571
{
565572
BLOG_D("TSubDomainManip(" << Tenant->Path << ") alter subdomain version " << Version);
566-
567-
auto request = MakeHolder<TEvTxUserProxy::TEvProposeTransaction>();
568-
request->Record.SetDatabaseName(TString(ExtractDomain(Subdomain.first)));
569-
request->Record.SetExecTimeoutPeriod(Max<ui64>());
570-
571-
if (Tenant->UserToken.GetUserSID())
572-
request->Record.SetUserToken(Tenant->UserToken.SerializeAsString());
573+
auto request = MakeProposeTransaction();
573574

574575
auto &tx = *request->Record.MutableTransaction()->MutableModifyScheme();
575576
tx.SetWorkingDir(Subdomain.first);
@@ -591,13 +592,7 @@ class TSubDomainManip : public TActorBootstrapped<TSubDomainManip> {
591592
void CreateSubdomain(const TActorContext &ctx)
592593
{
593594
BLOG_D("TSubDomainManip(" << Tenant->Path << ") create subdomain");
594-
595-
auto request = MakeHolder<TEvTxUserProxy::TEvProposeTransaction>();
596-
request->Record.SetDatabaseName(TString(ExtractDomain(Subdomain.first)));
597-
request->Record.SetExecTimeoutPeriod(Max<ui64>());
598-
599-
if (Tenant->UserToken.GetUserSID())
600-
request->Record.SetUserToken(Tenant->UserToken.SerializeAsString());
595+
auto request = MakeProposeTransaction();
601596

602597
auto &tx = *request->Record.MutableTransaction()->MutableModifyScheme();
603598
tx.SetWorkingDir(Subdomain.first);
@@ -622,12 +617,8 @@ class TSubDomainManip : public TActorBootstrapped<TSubDomainManip> {
622617
void DropSubdomain(const TActorContext &ctx)
623618
{
624619
BLOG_D("TSubDomainManip(" << Tenant->Path << ") drop subdomain");
620+
auto request = MakeProposeTransaction();
625621

626-
auto request = MakeHolder<TEvTxUserProxy::TEvProposeTransaction>();
627-
request->Record.SetDatabaseName(TString(ExtractDomain(Subdomain.first)));
628-
request->Record.SetExecTimeoutPeriod(Max<ui64>());
629-
if (Tenant->UserToken.GetUserSID())
630-
request->Record.SetUserToken(Tenant->UserToken.SerializeAsString());
631622
auto &tx = *request->Record.MutableTransaction()->MutableModifyScheme();
632623
if (Tenant->IsExternalSubdomain) {
633624
tx.SetOperationType(NKikimrSchemeOp::ESchemeOpForceDropExtSubDomain);
@@ -944,7 +935,7 @@ class TScaleRecommenderManip : public TActorBootstrapped<TScaleRecommenderManip>
944935

945936
Become(&TThis::StateResolveHive);
946937
ResolveHive(ctx);
947-
}
938+
}
948939

949940
void ResolveHive(const TActorContext &ctx) const {
950941
auto request = MakeHolder<NSchemeCache::TSchemeCacheNavigate>();
@@ -1004,7 +995,7 @@ class TScaleRecommenderManip : public TActorBootstrapped<TScaleRecommenderManip>
1004995
if (!domainInfo || !domainInfo->Params.HasHive()) {
1005996
LOG_ERROR_S(ctx, NKikimrServices::CMS_TENANTS,
1006997
"TScaleRecommenderManip resolved tenant "
1007-
<< Tenant->Path
998+
<< Tenant->Path
1008999
<< " that has no hive"
10091000
<< ", entry# " << entry.ToString());
10101001
Finish();
@@ -1035,7 +1026,7 @@ class TScaleRecommenderManip : public TActorBootstrapped<TScaleRecommenderManip>
10351026
default:
10361027
LOG_ERROR_S(ctx, NKikimrServices::CMS_TENANTS,
10371028
"TScaleRecommenderManip got unknown taget for target tracking policy for "
1038-
<< Tenant->Path
1029+
<< Tenant->Path
10391030
<< ", policy# " << p.target_tracking_policy().ShortDebugString());
10401031
Finish();
10411032
break;
@@ -1045,7 +1036,7 @@ class TScaleRecommenderManip : public TActorBootstrapped<TScaleRecommenderManip>
10451036
default:
10461037
LOG_ERROR_S(ctx, NKikimrServices::CMS_TENANTS,
10471038
"TScaleRecommenderManip got unknown scale policy for "
1048-
<< Tenant->Path
1039+
<< Tenant->Path
10491040
<< ", policies# " << Tenant->ScaleRecommenderPolicies->ShortDebugString());
10501041
Finish();
10511042
return;
@@ -1102,7 +1093,7 @@ class TScaleRecommenderManip : public TActorBootstrapped<TScaleRecommenderManip>
11021093
case NKikimrProto::UNKNOWN:
11031094
LOG_ERROR_S(ctx, NKikimrServices::CMS_TENANTS,
11041095
"TScaleRecommenderManip got error reply during configuring hive for "
1105-
<< Tenant->Path
1096+
<< Tenant->Path
11061097
<< ", reply# " << ev->Get()->Record.ShortDebugString());
11071098
Finish();
11081099
break;

‎ydb/core/grpc_services/base/base.h‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -448,6 +448,8 @@ class IRequestProxyCtx
448448
}
449449
virtual void SetAuditLogHook(TAuditLogHook&& hook) = 0;
450450
virtual void SetDiskQuotaExceeded(bool disk) = 0;
451+
452+
virtual TString GetRpcMethodName() const = 0;
451453
};
452454

453455
// Request context
@@ -700,6 +702,10 @@ class TRefreshTokenImpl
700702
Y_ABORT("unimplemented for TRefreshTokenImpl");
701703
}
702704

705+
TString GetRpcMethodName() const override {
706+
return {};
707+
}
708+
703709
// IRequestCtxBase
704710
//
705711
void AddAuditLogPart(const TStringBuf&, const TString&) override {
@@ -1003,6 +1009,10 @@ class TGRpcRequestBiStreamWrapper
10031009
return Ctx_->WriteAndFinish(std::move(message), options, grpcStatus);
10041010
}
10051011

1012+
TString GetRpcMethodName() const override {
1013+
return Ctx_->GetRpcMethodName();
1014+
}
1015+
10061016
private:
10071017
TIntrusivePtr<IStreamCtx> Ctx_;
10081018
TIntrusiveConstPtr<NACLib::TUserToken> InternalToken_;
@@ -1716,12 +1726,13 @@ class TEvRequestAuthAndCheck
17161726
: public IRequestProxyCtx
17171727
, public TEventLocal<TEvRequestAuthAndCheck, TRpcServices::EvRequestAuthAndCheck> {
17181728
public:
1719-
TEvRequestAuthAndCheck(const TString& database, const TMaybe<TString>& ydbToken, NActors::TActorId sender, TAuditMode auditMode)
1729+
TEvRequestAuthAndCheck(const TString& database, const TMaybe<TString>& ydbToken, NActors::TActorId sender, TAuditMode auditMode, TString peerName = {})
17201730
: Database(database)
17211731
, YdbToken(ydbToken)
17221732
, Sender(sender)
17231733
, AuthState(true)
17241734
, AuditMode(auditMode)
1735+
, PeerName(std::move(peerName))
17251736
{}
17261737

17271738
// IRequestProxyCtx
@@ -1875,7 +1886,7 @@ class TEvRequestAuthAndCheck
18751886
}
18761887

18771888
TString GetPeerName() const override {
1878-
return {};
1889+
return PeerName;
18791890
}
18801891

18811892
const TString& GetRequestName() const override {
@@ -1904,6 +1915,10 @@ class TEvRequestAuthAndCheck
19041915
return AuditMode;
19051916
}
19061917

1918+
TString GetRpcMethodName() const override {
1919+
return {};
1920+
}
1921+
19071922
TString Database;
19081923
TMaybe<TString> YdbToken;
19091924
NActors::TActorId Sender;
@@ -1916,6 +1931,7 @@ class TEvRequestAuthAndCheck
19161931
TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
19171932
TInstant deadline = TInstant::Now() + TDuration::Seconds(10);
19181933
TAuditMode AuditMode;
1934+
TString PeerName;
19191935

19201936
inline static const TString EmptySerializedTokenMessage;
19211937
};

‎ydb/core/grpc_services/grpc_request_check_actor.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -480,8 +480,8 @@ class TGrpcRequestCheckActor
480480

481481
if (auditEnabledReceived || auditEnabledCompleted) {
482482
if constexpr (IsGrpcRequest) {
483-
if (const auto* reqCtx = dynamic_cast<const IRequestCtx*>(requestBaseCtx)) {
484-
requestBaseCtx->AddAuditLogPart("grpc_method", reqCtx->GetRpcMethodName());
483+
if (TString grpcMethod = requestBaseCtx->GetRpcMethodName()) {
484+
requestBaseCtx->AddAuditLogPart("grpc_method", requestBaseCtx->GetRpcMethodName());
485485
}
486486
}
487487
const TString sanitizedToken = TBase::GetSanitizedToken();

‎ydb/core/grpc_streaming/grpc_streaming.h‎

Lines changed: 42 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -17,16 +17,8 @@
1717
namespace NKikimr {
1818
namespace NGRpcServer {
1919

20-
/**
21-
* Context for performing operations on a bidirectional stream
22-
*
23-
* Only one thread is allowed to call methods on this class at any time
24-
*/
25-
template<class TIn, class TOut>
26-
class IGRpcStreamingContext : public TThrRefBase {
20+
class IGRpcStreamingContextBase : public TThrRefBase {
2721
public:
28-
using ISelf = IGRpcStreamingContext<TIn, TOut>;
29-
3022
enum EEv {
3123
EvBegin = EventSpaceBegin(TKikimrEvents::ES_GRPC_STREAMING),
3224

@@ -35,11 +27,6 @@ class IGRpcStreamingContext : public TThrRefBase {
3527
EvNotifiedWhenDone,
3628
};
3729

38-
struct TEvReadFinished : public TEventLocal<TEvReadFinished, EvReadFinished> {
39-
TIn Record;
40-
bool Success;
41-
};
42-
4330
struct TEvWriteFinished : public TEventLocal<TEvWriteFinished, EvWriteFinished> {
4431
bool Success;
4532
};
@@ -53,9 +40,8 @@ class IGRpcStreamingContext : public TThrRefBase {
5340
};
5441

5542
public:
56-
virtual ~IGRpcStreamingContext() = default;
43+
virtual ~IGRpcStreamingContextBase() = default;
5744

58-
public:
5945
/**
6046
* Asynchronously cancels the request
6147
*
@@ -79,6 +65,38 @@ class IGRpcStreamingContext : public TThrRefBase {
7965
*/
8066
virtual bool Read() = 0;
8167

68+
/**
69+
* Schedules stream termination with the specified status
70+
*
71+
* Only the first call is accepted, after which new Read or Write calls
72+
* are no longer permitted and ignored.
73+
*/
74+
virtual bool Finish(const grpc::Status& status) = 0;
75+
76+
virtual NYdbGrpc::TAuthState& GetAuthState() const = 0;
77+
virtual TString GetPeerName() const = 0;
78+
virtual TVector<TStringBuf> GetPeerMetaValues(TStringBuf key) const = 0;
79+
virtual grpc_compression_level GetCompressionLevel() const = 0;
80+
virtual void UseDatabase(const TString& database) = 0;
81+
virtual TString GetRpcMethodName() const = 0;
82+
};
83+
84+
/**
85+
* Context for performing operations on a bidirectional stream
86+
*
87+
* Only one thread is allowed to call methods on this class at any time
88+
*/
89+
template<class TIn, class TOut>
90+
class IGRpcStreamingContext : public IGRpcStreamingContextBase {
91+
public:
92+
using ISelf = IGRpcStreamingContext<TIn, TOut>;
93+
94+
struct TEvReadFinished : public TEventLocal<TEvReadFinished, EvReadFinished> {
95+
TIn Record;
96+
bool Success;
97+
};
98+
99+
public:
82100
/**
83101
* Schedules the next message write
84102
*
@@ -88,14 +106,6 @@ class IGRpcStreamingContext : public TThrRefBase {
88106
*/
89107
virtual bool Write(TOut&& message, const grpc::WriteOptions& options = { }) = 0;
90108

91-
/**
92-
* Schedules stream termination with the specified status
93-
*
94-
* Only the first call is accepted, after which new Read or Write calls
95-
* are no longer permitted and ignored.
96-
*/
97-
virtual bool Finish(const grpc::Status& status) = 0;
98-
99109
/**
100110
* Schedules the next message write combined with the status
101111
*
@@ -109,13 +119,6 @@ class IGRpcStreamingContext : public TThrRefBase {
109119
* This is similar to Write and Finish combined into a more efficient call.
110120
*/
111121
virtual bool WriteAndFinish(TOut&& message, const grpc::WriteOptions& options, const grpc::Status& status) = 0;
112-
113-
public:
114-
virtual NYdbGrpc::TAuthState& GetAuthState() const = 0;
115-
virtual TString GetPeerName() const = 0;
116-
virtual TVector<TStringBuf> GetPeerMetaValues(TStringBuf key) const = 0;
117-
virtual grpc_compression_level GetCompressionLevel() const = 0;
118-
virtual void UseDatabase(const TString& database) = 0;
119122
};
120123

121124
template<class TIn, class TOut, class TServer, int LoggerServiceId>
@@ -602,6 +605,10 @@ class TGRpcStreamingRequest final
602605
}
603606
}
604607

608+
TString GetRpcMethodName() const {
609+
return TStringBuilder() << TServer::TCurrentGRpcService::service_full_name() << '/' << Name;
610+
}
611+
605612
private:
606613
class TFacade : public IContext {
607614
public:
@@ -661,6 +668,10 @@ class TGRpcStreamingRequest final
661668
Self->UseDatabase(database);
662669
}
663670

671+
TString GetRpcMethodName() const override {
672+
return Self->GetRpcMethodName();
673+
}
674+
664675
private:
665676
TIntrusivePtr<TSelf> Self;
666677
};

0 commit comments

Comments
 (0)