From eaff973d2992c2dc71bc1d091e3f27987addbfe8 Mon Sep 17 00:00:00 2001 From: Benny Zlotnik Date: Thu, 3 Sep 2026 18:39:40 +0300 Subject: [PATCH 1/2] feat: have exporter request a new token when about to expire - Add RotateToken RPC to ControllerService in protocol/proto/jumpstarter/v1/jumpstarter.proto. - Implement RotateToken in ControllerService to generate a new signed token and update the exporter Secret. - Regenerate jumpstarter.v1 protobuf bindings for Go and Python. - Add token_refresh helper to decode JWT claims and calculate renewal lead times and sleep intervals. - Add _token_refresh_loop in Exporter to automatically request a new token before expiry. - Wire on_token_rotated callback in ExporterConfigV1Alpha1 to update in-memory state and persist to disk. - Add comprehensive unit tests in controller and python packages. --- .../protocol/jumpstarter/v1/jumpstarter.pb.go | 262 ++++++++++++------ .../jumpstarter/v1/jumpstarter_grpc.pb.go | 40 +++ .../internal/service/controller_service.go | 59 ++++ .../service/controller_service_test.go | 116 ++++++++ .../proto/jumpstarter/v1/jumpstarter.proto | 12 + .../jumpstarter/v1/jumpstarter_pb2.py | 14 +- .../jumpstarter/v1/jumpstarter_pb2.pyi | 37 +++ .../jumpstarter/v1/jumpstarter_pb2_grpc.py | 44 +++ .../jumpstarter/v1/jumpstarter_pb2_grpc.pyi | 41 ++- .../jumpstarter/config/exporter.py | 22 +- .../jumpstarter/exporter/exporter.py | 157 +++++++++-- .../exporter/exporter_token_refresh_test.py | 245 ++++++++++++++++ .../jumpstarter/exporter/token_refresh.py | 132 +++++++++ .../exporter/token_refresh_test.py | 120 ++++++++ 14 files changed, 1186 insertions(+), 115 deletions(-) create mode 100644 python/packages/jumpstarter/jumpstarter/exporter/exporter_token_refresh_test.py create mode 100644 python/packages/jumpstarter/jumpstarter/exporter/token_refresh.py create mode 100644 python/packages/jumpstarter/jumpstarter/exporter/token_refresh_test.py diff --git a/controller/internal/protocol/jumpstarter/v1/jumpstarter.pb.go b/controller/internal/protocol/jumpstarter/v1/jumpstarter.pb.go index 23ea3296c..c8c836539 100644 --- a/controller/internal/protocol/jumpstarter/v1/jumpstarter.pb.go +++ b/controller/internal/protocol/jumpstarter/v1/jumpstarter.pb.go @@ -1965,6 +1965,96 @@ func (x *TelemetryEndpoint) GetMinSeverity() string { return "" } +// Request to rotate the authentication token for an exporter. +type RotateTokenRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RotateTokenRequest) Reset() { + *x = RotateTokenRequest{} + mi := &file_jumpstarter_v1_jumpstarter_proto_msgTypes[37] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RotateTokenRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RotateTokenRequest) ProtoMessage() {} + +func (x *RotateTokenRequest) ProtoReflect() protoreflect.Message { + mi := &file_jumpstarter_v1_jumpstarter_proto_msgTypes[37] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RotateTokenRequest.ProtoReflect.Descriptor instead. +func (*RotateTokenRequest) Descriptor() ([]byte, []int) { + return file_jumpstarter_v1_jumpstarter_proto_rawDescGZIP(), []int{37} +} + +// Response containing the newly rotated token. +type RotateTokenResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Token string `protobuf:"bytes,1,opt,name=token,proto3" json:"token,omitempty"` // The new authentication token. + Expiry *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=expiry,proto3" json:"expiry,omitempty"` // The expiration time of the new token. + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RotateTokenResponse) Reset() { + *x = RotateTokenResponse{} + mi := &file_jumpstarter_v1_jumpstarter_proto_msgTypes[38] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RotateTokenResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RotateTokenResponse) ProtoMessage() {} + +func (x *RotateTokenResponse) ProtoReflect() protoreflect.Message { + mi := &file_jumpstarter_v1_jumpstarter_proto_msgTypes[38] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RotateTokenResponse.ProtoReflect.Descriptor instead. +func (*RotateTokenResponse) Descriptor() ([]byte, []int) { + return file_jumpstarter_v1_jumpstarter_proto_rawDescGZIP(), []int{38} +} + +func (x *RotateTokenResponse) GetToken() string { + if x != nil { + return x.Token + } + return "" +} + +func (x *RotateTokenResponse) GetExpiry() *timestamppb.Timestamp { + if x != nil { + return x.Expiry + } + return nil +} + var File_jumpstarter_v1_jumpstarter_proto protoreflect.FileDescriptor const file_jumpstarter_v1_jumpstarter_proto_rawDesc = "" + @@ -2125,7 +2215,11 @@ const file_jumpstarter_v1_jumpstarter_proto_rawDesc = "" + "\x11TelemetryEndpoint\x12\x1a\n" + "\bendpoint\x18\x01 \x01(\tR\bendpoint\x12 \n" + "\vcertificate\x18\x02 \x01(\tR\vcertificate\x12!\n" + - "\fmin_severity\x18\x03 \x01(\tR\vminSeverity2\xb5\a\n" + + "\fmin_severity\x18\x03 \x01(\tR\vminSeverity\"\x14\n" + + "\x12RotateTokenRequest\"_\n" + + "\x13RotateTokenResponse\x12\x14\n" + + "\x05token\x18\x01 \x01(\tR\x05token\x122\n" + + "\x06expiry\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\x06expiry2\x8d\b\n" + "\x11ControllerService\x12M\n" + "\bRegister\x12\x1f.jumpstarter.v1.RegisterRequest\x1a .jumpstarter.v1.RegisterResponse\x12S\n" + "\n" + @@ -2139,7 +2233,8 @@ const file_jumpstarter_v1_jumpstarter_proto_rawDesc = "" + "\fReleaseLease\x12#.jumpstarter.v1.ReleaseLeaseRequest\x1a$.jumpstarter.v1.ReleaseLeaseResponse\x12S\n" + "\n" + "ListLeases\x12!.jumpstarter.v1.ListLeasesRequest\x1a\".jumpstarter.v1.ListLeasesResponse\x12n\n" + - "\x13GetServiceEndpoints\x12*.jumpstarter.v1.GetServiceEndpointsRequest\x1a+.jumpstarter.v1.GetServiceEndpointsResponse2\xd7\x04\n" + + "\x13GetServiceEndpoints\x12*.jumpstarter.v1.GetServiceEndpointsRequest\x1a+.jumpstarter.v1.GetServiceEndpointsResponse\x12V\n" + + "\vRotateToken\x12\".jumpstarter.v1.RotateTokenRequest\x1a#.jumpstarter.v1.RotateTokenResponse2\xd7\x04\n" + "\x0fExporterService\x12F\n" + "\tGetReport\x12\x16.google.protobuf.Empty\x1a!.jumpstarter.v1.GetReportResponse\x12S\n" + "\n" + @@ -2164,7 +2259,7 @@ func file_jumpstarter_v1_jumpstarter_proto_rawDescGZIP() []byte { return file_jumpstarter_v1_jumpstarter_proto_rawDescData } -var file_jumpstarter_v1_jumpstarter_proto_msgTypes = make([]protoimpl.MessageInfo, 43) +var file_jumpstarter_v1_jumpstarter_proto_msgTypes = make([]protoimpl.MessageInfo, 45) var file_jumpstarter_v1_jumpstarter_proto_goTypes = []any{ (*RegisterRequest)(nil), // 0: jumpstarter.v1.RegisterRequest (*DriverInstanceReport)(nil), // 1: jumpstarter.v1.DriverInstanceReport @@ -2203,89 +2298,94 @@ var file_jumpstarter_v1_jumpstarter_proto_goTypes = []any{ (*GetServiceEndpointsRequest)(nil), // 34: jumpstarter.v1.GetServiceEndpointsRequest (*GetServiceEndpointsResponse)(nil), // 35: jumpstarter.v1.GetServiceEndpointsResponse (*TelemetryEndpoint)(nil), // 36: jumpstarter.v1.TelemetryEndpoint - nil, // 37: jumpstarter.v1.RegisterRequest.LabelsEntry - nil, // 38: jumpstarter.v1.DriverInstanceReport.LabelsEntry - nil, // 39: jumpstarter.v1.DriverInstanceReport.MethodsDescriptionEntry - nil, // 40: jumpstarter.v1.StatusResponse.ContextEntry - nil, // 41: jumpstarter.v1.GetReportResponse.LabelsEntry - nil, // 42: jumpstarter.v1.LogStreamResponse.StructuredFieldsEntry - (ExporterStatus)(0), // 43: jumpstarter.v1.ExporterStatus - (*structpb.Value)(nil), // 44: google.protobuf.Value - (LogSource)(0), // 45: jumpstarter.v1.LogSource - (*timestamppb.Timestamp)(nil), // 46: google.protobuf.Timestamp - (*durationpb.Duration)(nil), // 47: google.protobuf.Duration - (*LabelSelector)(nil), // 48: jumpstarter.v1.LabelSelector - (*Condition)(nil), // 49: jumpstarter.v1.Condition - (*emptypb.Empty)(nil), // 50: google.protobuf.Empty + (*RotateTokenRequest)(nil), // 37: jumpstarter.v1.RotateTokenRequest + (*RotateTokenResponse)(nil), // 38: jumpstarter.v1.RotateTokenResponse + nil, // 39: jumpstarter.v1.RegisterRequest.LabelsEntry + nil, // 40: jumpstarter.v1.DriverInstanceReport.LabelsEntry + nil, // 41: jumpstarter.v1.DriverInstanceReport.MethodsDescriptionEntry + nil, // 42: jumpstarter.v1.StatusResponse.ContextEntry + nil, // 43: jumpstarter.v1.GetReportResponse.LabelsEntry + nil, // 44: jumpstarter.v1.LogStreamResponse.StructuredFieldsEntry + (ExporterStatus)(0), // 45: jumpstarter.v1.ExporterStatus + (*structpb.Value)(nil), // 46: google.protobuf.Value + (LogSource)(0), // 47: jumpstarter.v1.LogSource + (*timestamppb.Timestamp)(nil), // 48: google.protobuf.Timestamp + (*durationpb.Duration)(nil), // 49: google.protobuf.Duration + (*LabelSelector)(nil), // 50: jumpstarter.v1.LabelSelector + (*Condition)(nil), // 51: jumpstarter.v1.Condition + (*emptypb.Empty)(nil), // 52: google.protobuf.Empty } var file_jumpstarter_v1_jumpstarter_proto_depIdxs = []int32{ - 37, // 0: jumpstarter.v1.RegisterRequest.labels:type_name -> jumpstarter.v1.RegisterRequest.LabelsEntry + 39, // 0: jumpstarter.v1.RegisterRequest.labels:type_name -> jumpstarter.v1.RegisterRequest.LabelsEntry 1, // 1: jumpstarter.v1.RegisterRequest.reports:type_name -> jumpstarter.v1.DriverInstanceReport - 38, // 2: jumpstarter.v1.DriverInstanceReport.labels:type_name -> jumpstarter.v1.DriverInstanceReport.LabelsEntry - 39, // 3: jumpstarter.v1.DriverInstanceReport.methods_description:type_name -> jumpstarter.v1.DriverInstanceReport.MethodsDescriptionEntry - 40, // 4: jumpstarter.v1.StatusResponse.context:type_name -> jumpstarter.v1.StatusResponse.ContextEntry - 43, // 5: jumpstarter.v1.ReportStatusRequest.status:type_name -> jumpstarter.v1.ExporterStatus - 41, // 6: jumpstarter.v1.GetReportResponse.labels:type_name -> jumpstarter.v1.GetReportResponse.LabelsEntry + 40, // 2: jumpstarter.v1.DriverInstanceReport.labels:type_name -> jumpstarter.v1.DriverInstanceReport.LabelsEntry + 41, // 3: jumpstarter.v1.DriverInstanceReport.methods_description:type_name -> jumpstarter.v1.DriverInstanceReport.MethodsDescriptionEntry + 42, // 4: jumpstarter.v1.StatusResponse.context:type_name -> jumpstarter.v1.StatusResponse.ContextEntry + 45, // 5: jumpstarter.v1.ReportStatusRequest.status:type_name -> jumpstarter.v1.ExporterStatus + 43, // 6: jumpstarter.v1.GetReportResponse.labels:type_name -> jumpstarter.v1.GetReportResponse.LabelsEntry 1, // 7: jumpstarter.v1.GetReportResponse.reports:type_name -> jumpstarter.v1.DriverInstanceReport 14, // 8: jumpstarter.v1.GetReportResponse.alternative_endpoints:type_name -> jumpstarter.v1.Endpoint - 44, // 9: jumpstarter.v1.DriverCallRequest.args:type_name -> google.protobuf.Value - 44, // 10: jumpstarter.v1.DriverCallResponse.result:type_name -> google.protobuf.Value - 44, // 11: jumpstarter.v1.StreamingDriverCallRequest.args:type_name -> google.protobuf.Value - 44, // 12: jumpstarter.v1.StreamingDriverCallResponse.result:type_name -> google.protobuf.Value - 45, // 13: jumpstarter.v1.LogStreamResponse.source:type_name -> jumpstarter.v1.LogSource - 46, // 14: jumpstarter.v1.LogStreamResponse.timestamp:type_name -> google.protobuf.Timestamp - 42, // 15: jumpstarter.v1.LogStreamResponse.structured_fields:type_name -> jumpstarter.v1.LogStreamResponse.StructuredFieldsEntry - 47, // 16: jumpstarter.v1.GetLeaseResponse.duration:type_name -> google.protobuf.Duration - 48, // 17: jumpstarter.v1.GetLeaseResponse.selector:type_name -> jumpstarter.v1.LabelSelector - 46, // 18: jumpstarter.v1.GetLeaseResponse.begin_time:type_name -> google.protobuf.Timestamp - 46, // 19: jumpstarter.v1.GetLeaseResponse.end_time:type_name -> google.protobuf.Timestamp - 49, // 20: jumpstarter.v1.GetLeaseResponse.conditions:type_name -> jumpstarter.v1.Condition - 47, // 21: jumpstarter.v1.RequestLeaseRequest.duration:type_name -> google.protobuf.Duration - 48, // 22: jumpstarter.v1.RequestLeaseRequest.selector:type_name -> jumpstarter.v1.LabelSelector - 43, // 23: jumpstarter.v1.GetStatusResponse.status:type_name -> jumpstarter.v1.ExporterStatus - 43, // 24: jumpstarter.v1.GetStatusResponse.previous_status:type_name -> jumpstarter.v1.ExporterStatus + 46, // 9: jumpstarter.v1.DriverCallRequest.args:type_name -> google.protobuf.Value + 46, // 10: jumpstarter.v1.DriverCallResponse.result:type_name -> google.protobuf.Value + 46, // 11: jumpstarter.v1.StreamingDriverCallRequest.args:type_name -> google.protobuf.Value + 46, // 12: jumpstarter.v1.StreamingDriverCallResponse.result:type_name -> google.protobuf.Value + 47, // 13: jumpstarter.v1.LogStreamResponse.source:type_name -> jumpstarter.v1.LogSource + 48, // 14: jumpstarter.v1.LogStreamResponse.timestamp:type_name -> google.protobuf.Timestamp + 44, // 15: jumpstarter.v1.LogStreamResponse.structured_fields:type_name -> jumpstarter.v1.LogStreamResponse.StructuredFieldsEntry + 49, // 16: jumpstarter.v1.GetLeaseResponse.duration:type_name -> google.protobuf.Duration + 50, // 17: jumpstarter.v1.GetLeaseResponse.selector:type_name -> jumpstarter.v1.LabelSelector + 48, // 18: jumpstarter.v1.GetLeaseResponse.begin_time:type_name -> google.protobuf.Timestamp + 48, // 19: jumpstarter.v1.GetLeaseResponse.end_time:type_name -> google.protobuf.Timestamp + 51, // 20: jumpstarter.v1.GetLeaseResponse.conditions:type_name -> jumpstarter.v1.Condition + 49, // 21: jumpstarter.v1.RequestLeaseRequest.duration:type_name -> google.protobuf.Duration + 50, // 22: jumpstarter.v1.RequestLeaseRequest.selector:type_name -> jumpstarter.v1.LabelSelector + 45, // 23: jumpstarter.v1.GetStatusResponse.status:type_name -> jumpstarter.v1.ExporterStatus + 45, // 24: jumpstarter.v1.GetStatusResponse.previous_status:type_name -> jumpstarter.v1.ExporterStatus 36, // 25: jumpstarter.v1.GetServiceEndpointsResponse.telemetry_endpoints:type_name -> jumpstarter.v1.TelemetryEndpoint - 0, // 26: jumpstarter.v1.ControllerService.Register:input_type -> jumpstarter.v1.RegisterRequest - 3, // 27: jumpstarter.v1.ControllerService.Unregister:input_type -> jumpstarter.v1.UnregisterRequest - 11, // 28: jumpstarter.v1.ControllerService.ReportStatus:input_type -> jumpstarter.v1.ReportStatusRequest - 5, // 29: jumpstarter.v1.ControllerService.Listen:input_type -> jumpstarter.v1.ListenRequest - 7, // 30: jumpstarter.v1.ControllerService.Status:input_type -> jumpstarter.v1.StatusRequest - 9, // 31: jumpstarter.v1.ControllerService.Dial:input_type -> jumpstarter.v1.DialRequest - 22, // 32: jumpstarter.v1.ControllerService.GetLease:input_type -> jumpstarter.v1.GetLeaseRequest - 24, // 33: jumpstarter.v1.ControllerService.RequestLease:input_type -> jumpstarter.v1.RequestLeaseRequest - 26, // 34: jumpstarter.v1.ControllerService.ReleaseLease:input_type -> jumpstarter.v1.ReleaseLeaseRequest - 28, // 35: jumpstarter.v1.ControllerService.ListLeases:input_type -> jumpstarter.v1.ListLeasesRequest - 34, // 36: jumpstarter.v1.ControllerService.GetServiceEndpoints:input_type -> jumpstarter.v1.GetServiceEndpointsRequest - 50, // 37: jumpstarter.v1.ExporterService.GetReport:input_type -> google.protobuf.Empty - 15, // 38: jumpstarter.v1.ExporterService.DriverCall:input_type -> jumpstarter.v1.DriverCallRequest - 17, // 39: jumpstarter.v1.ExporterService.StreamingDriverCall:input_type -> jumpstarter.v1.StreamingDriverCallRequest - 50, // 40: jumpstarter.v1.ExporterService.LogStream:input_type -> google.protobuf.Empty - 20, // 41: jumpstarter.v1.ExporterService.Reset:input_type -> jumpstarter.v1.ResetRequest - 30, // 42: jumpstarter.v1.ExporterService.GetStatus:input_type -> jumpstarter.v1.GetStatusRequest - 32, // 43: jumpstarter.v1.ExporterService.EndSession:input_type -> jumpstarter.v1.EndSessionRequest - 2, // 44: jumpstarter.v1.ControllerService.Register:output_type -> jumpstarter.v1.RegisterResponse - 4, // 45: jumpstarter.v1.ControllerService.Unregister:output_type -> jumpstarter.v1.UnregisterResponse - 12, // 46: jumpstarter.v1.ControllerService.ReportStatus:output_type -> jumpstarter.v1.ReportStatusResponse - 6, // 47: jumpstarter.v1.ControllerService.Listen:output_type -> jumpstarter.v1.ListenResponse - 8, // 48: jumpstarter.v1.ControllerService.Status:output_type -> jumpstarter.v1.StatusResponse - 10, // 49: jumpstarter.v1.ControllerService.Dial:output_type -> jumpstarter.v1.DialResponse - 23, // 50: jumpstarter.v1.ControllerService.GetLease:output_type -> jumpstarter.v1.GetLeaseResponse - 25, // 51: jumpstarter.v1.ControllerService.RequestLease:output_type -> jumpstarter.v1.RequestLeaseResponse - 27, // 52: jumpstarter.v1.ControllerService.ReleaseLease:output_type -> jumpstarter.v1.ReleaseLeaseResponse - 29, // 53: jumpstarter.v1.ControllerService.ListLeases:output_type -> jumpstarter.v1.ListLeasesResponse - 35, // 54: jumpstarter.v1.ControllerService.GetServiceEndpoints:output_type -> jumpstarter.v1.GetServiceEndpointsResponse - 13, // 55: jumpstarter.v1.ExporterService.GetReport:output_type -> jumpstarter.v1.GetReportResponse - 16, // 56: jumpstarter.v1.ExporterService.DriverCall:output_type -> jumpstarter.v1.DriverCallResponse - 18, // 57: jumpstarter.v1.ExporterService.StreamingDriverCall:output_type -> jumpstarter.v1.StreamingDriverCallResponse - 19, // 58: jumpstarter.v1.ExporterService.LogStream:output_type -> jumpstarter.v1.LogStreamResponse - 21, // 59: jumpstarter.v1.ExporterService.Reset:output_type -> jumpstarter.v1.ResetResponse - 31, // 60: jumpstarter.v1.ExporterService.GetStatus:output_type -> jumpstarter.v1.GetStatusResponse - 33, // 61: jumpstarter.v1.ExporterService.EndSession:output_type -> jumpstarter.v1.EndSessionResponse - 44, // [44:62] is the sub-list for method output_type - 26, // [26:44] is the sub-list for method input_type - 26, // [26:26] is the sub-list for extension type_name - 26, // [26:26] is the sub-list for extension extendee - 0, // [0:26] is the sub-list for field type_name + 48, // 26: jumpstarter.v1.RotateTokenResponse.expiry:type_name -> google.protobuf.Timestamp + 0, // 27: jumpstarter.v1.ControllerService.Register:input_type -> jumpstarter.v1.RegisterRequest + 3, // 28: jumpstarter.v1.ControllerService.Unregister:input_type -> jumpstarter.v1.UnregisterRequest + 11, // 29: jumpstarter.v1.ControllerService.ReportStatus:input_type -> jumpstarter.v1.ReportStatusRequest + 5, // 30: jumpstarter.v1.ControllerService.Listen:input_type -> jumpstarter.v1.ListenRequest + 7, // 31: jumpstarter.v1.ControllerService.Status:input_type -> jumpstarter.v1.StatusRequest + 9, // 32: jumpstarter.v1.ControllerService.Dial:input_type -> jumpstarter.v1.DialRequest + 22, // 33: jumpstarter.v1.ControllerService.GetLease:input_type -> jumpstarter.v1.GetLeaseRequest + 24, // 34: jumpstarter.v1.ControllerService.RequestLease:input_type -> jumpstarter.v1.RequestLeaseRequest + 26, // 35: jumpstarter.v1.ControllerService.ReleaseLease:input_type -> jumpstarter.v1.ReleaseLeaseRequest + 28, // 36: jumpstarter.v1.ControllerService.ListLeases:input_type -> jumpstarter.v1.ListLeasesRequest + 34, // 37: jumpstarter.v1.ControllerService.GetServiceEndpoints:input_type -> jumpstarter.v1.GetServiceEndpointsRequest + 37, // 38: jumpstarter.v1.ControllerService.RotateToken:input_type -> jumpstarter.v1.RotateTokenRequest + 52, // 39: jumpstarter.v1.ExporterService.GetReport:input_type -> google.protobuf.Empty + 15, // 40: jumpstarter.v1.ExporterService.DriverCall:input_type -> jumpstarter.v1.DriverCallRequest + 17, // 41: jumpstarter.v1.ExporterService.StreamingDriverCall:input_type -> jumpstarter.v1.StreamingDriverCallRequest + 52, // 42: jumpstarter.v1.ExporterService.LogStream:input_type -> google.protobuf.Empty + 20, // 43: jumpstarter.v1.ExporterService.Reset:input_type -> jumpstarter.v1.ResetRequest + 30, // 44: jumpstarter.v1.ExporterService.GetStatus:input_type -> jumpstarter.v1.GetStatusRequest + 32, // 45: jumpstarter.v1.ExporterService.EndSession:input_type -> jumpstarter.v1.EndSessionRequest + 2, // 46: jumpstarter.v1.ControllerService.Register:output_type -> jumpstarter.v1.RegisterResponse + 4, // 47: jumpstarter.v1.ControllerService.Unregister:output_type -> jumpstarter.v1.UnregisterResponse + 12, // 48: jumpstarter.v1.ControllerService.ReportStatus:output_type -> jumpstarter.v1.ReportStatusResponse + 6, // 49: jumpstarter.v1.ControllerService.Listen:output_type -> jumpstarter.v1.ListenResponse + 8, // 50: jumpstarter.v1.ControllerService.Status:output_type -> jumpstarter.v1.StatusResponse + 10, // 51: jumpstarter.v1.ControllerService.Dial:output_type -> jumpstarter.v1.DialResponse + 23, // 52: jumpstarter.v1.ControllerService.GetLease:output_type -> jumpstarter.v1.GetLeaseResponse + 25, // 53: jumpstarter.v1.ControllerService.RequestLease:output_type -> jumpstarter.v1.RequestLeaseResponse + 27, // 54: jumpstarter.v1.ControllerService.ReleaseLease:output_type -> jumpstarter.v1.ReleaseLeaseResponse + 29, // 55: jumpstarter.v1.ControllerService.ListLeases:output_type -> jumpstarter.v1.ListLeasesResponse + 35, // 56: jumpstarter.v1.ControllerService.GetServiceEndpoints:output_type -> jumpstarter.v1.GetServiceEndpointsResponse + 38, // 57: jumpstarter.v1.ControllerService.RotateToken:output_type -> jumpstarter.v1.RotateTokenResponse + 13, // 58: jumpstarter.v1.ExporterService.GetReport:output_type -> jumpstarter.v1.GetReportResponse + 16, // 59: jumpstarter.v1.ExporterService.DriverCall:output_type -> jumpstarter.v1.DriverCallResponse + 18, // 60: jumpstarter.v1.ExporterService.StreamingDriverCall:output_type -> jumpstarter.v1.StreamingDriverCallResponse + 19, // 61: jumpstarter.v1.ExporterService.LogStream:output_type -> jumpstarter.v1.LogStreamResponse + 21, // 62: jumpstarter.v1.ExporterService.Reset:output_type -> jumpstarter.v1.ResetResponse + 31, // 63: jumpstarter.v1.ExporterService.GetStatus:output_type -> jumpstarter.v1.GetStatusResponse + 33, // 64: jumpstarter.v1.ExporterService.EndSession:output_type -> jumpstarter.v1.EndSessionResponse + 46, // [46:65] is the sub-list for method output_type + 27, // [27:46] is the sub-list for method input_type + 27, // [27:27] is the sub-list for extension type_name + 27, // [27:27] is the sub-list for extension extendee + 0, // [0:27] is the sub-list for field type_name } func init() { file_jumpstarter_v1_jumpstarter_proto_init() } @@ -2308,7 +2408,7 @@ func file_jumpstarter_v1_jumpstarter_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_jumpstarter_v1_jumpstarter_proto_rawDesc), len(file_jumpstarter_v1_jumpstarter_proto_rawDesc)), NumEnums: 0, - NumMessages: 43, + NumMessages: 45, NumExtensions: 0, NumServices: 2, }, diff --git a/controller/internal/protocol/jumpstarter/v1/jumpstarter_grpc.pb.go b/controller/internal/protocol/jumpstarter/v1/jumpstarter_grpc.pb.go index 1692cd952..efc8898ae 100644 --- a/controller/internal/protocol/jumpstarter/v1/jumpstarter_grpc.pb.go +++ b/controller/internal/protocol/jumpstarter/v1/jumpstarter_grpc.pb.go @@ -33,6 +33,7 @@ const ( ControllerService_ReleaseLease_FullMethodName = "/jumpstarter.v1.ControllerService/ReleaseLease" ControllerService_ListLeases_FullMethodName = "/jumpstarter.v1.ControllerService/ListLeases" ControllerService_GetServiceEndpoints_FullMethodName = "/jumpstarter.v1.ControllerService/GetServiceEndpoints" + ControllerService_RotateToken_FullMethodName = "/jumpstarter.v1.ControllerService/RotateToken" ) // ControllerServiceClient is the client API for ControllerService service. @@ -70,6 +71,8 @@ type ControllerServiceClient interface { // Returns an empty list when no optional services are deployed. // Older controllers return UNIMPLEMENTED; callers must treat that as an empty list. GetServiceEndpoints(ctx context.Context, in *GetServiceEndpointsRequest, opts ...grpc.CallOption) (*GetServiceEndpointsResponse, error) + // Rotate the authentication token for the exporter. + RotateToken(ctx context.Context, in *RotateTokenRequest, opts ...grpc.CallOption) (*RotateTokenResponse, error) } type controllerServiceClient struct { @@ -208,6 +211,16 @@ func (c *controllerServiceClient) GetServiceEndpoints(ctx context.Context, in *G return out, nil } +func (c *controllerServiceClient) RotateToken(ctx context.Context, in *RotateTokenRequest, opts ...grpc.CallOption) (*RotateTokenResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(RotateTokenResponse) + err := c.cc.Invoke(ctx, ControllerService_RotateToken_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // ControllerServiceServer is the server API for ControllerService service. // All implementations must embed UnimplementedControllerServiceServer // for forward compatibility. @@ -243,6 +256,8 @@ type ControllerServiceServer interface { // Returns an empty list when no optional services are deployed. // Older controllers return UNIMPLEMENTED; callers must treat that as an empty list. GetServiceEndpoints(context.Context, *GetServiceEndpointsRequest) (*GetServiceEndpointsResponse, error) + // Rotate the authentication token for the exporter. + RotateToken(context.Context, *RotateTokenRequest) (*RotateTokenResponse, error) mustEmbedUnimplementedControllerServiceServer() } @@ -286,6 +301,9 @@ func (UnimplementedControllerServiceServer) ListLeases(context.Context, *ListLea func (UnimplementedControllerServiceServer) GetServiceEndpoints(context.Context, *GetServiceEndpointsRequest) (*GetServiceEndpointsResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetServiceEndpoints not implemented") } +func (UnimplementedControllerServiceServer) RotateToken(context.Context, *RotateTokenRequest) (*RotateTokenResponse, error) { + return nil, status.Error(codes.Unimplemented, "method RotateToken not implemented") +} func (UnimplementedControllerServiceServer) mustEmbedUnimplementedControllerServiceServer() {} func (UnimplementedControllerServiceServer) testEmbeddedByValue() {} @@ -491,6 +509,24 @@ func _ControllerService_GetServiceEndpoints_Handler(srv interface{}, ctx context return interceptor(ctx, in, info, handler) } +func _ControllerService_RotateToken_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RotateTokenRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ControllerServiceServer).RotateToken(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: ControllerService_RotateToken_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ControllerServiceServer).RotateToken(ctx, req.(*RotateTokenRequest)) + } + return interceptor(ctx, in, info, handler) +} + // ControllerService_ServiceDesc is the grpc.ServiceDesc for ControllerService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -534,6 +570,10 @@ var ControllerService_ServiceDesc = grpc.ServiceDesc{ MethodName: "GetServiceEndpoints", Handler: _ControllerService_GetServiceEndpoints_Handler, }, + { + MethodName: "RotateToken", + Handler: _ControllerService_RotateToken_Handler, + }, }, Streams: []grpc.StreamDesc{ { diff --git a/controller/internal/service/controller_service.go b/controller/internal/service/controller_service.go index bbbf0b841..df92eff79 100644 --- a/controller/internal/service/controller_service.go +++ b/controller/internal/service/controller_service.go @@ -342,6 +342,65 @@ func (s *ControllerService) GetServiceEndpoints( return resp, nil } +// RotateToken signs a new token for the authenticated exporter, updates the +// corresponding Secret in Kubernetes, and returns the new token and its expiry. +func (s *ControllerService) RotateToken( + ctx context.Context, + req *pb.RotateTokenRequest, +) (*pb.RotateTokenResponse, error) { + exporter, err := s.authenticateExporter(ctx) + if err != nil { + return nil, err + } + + if s.Signer == nil { + return nil, status.Errorf(codes.FailedPrecondition, "token signer not configured") + } + + token, err := s.Signer.Token(exporter.InternalSubject()) + if err != nil { + return nil, status.Errorf(codes.Internal, "failed to sign token: %s", err) + } + + secretName := exporter.Name + "-exporter" + var secret corev1.Secret + if err := s.Client.Get(ctx, types.NamespacedName{ + Namespace: exporter.Namespace, + Name: secretName, + }, &secret); err != nil { + return nil, status.Errorf(codes.Internal, "failed to get credential secret: %s", err) + } + + original := client.MergeFrom(secret.DeepCopy()) + if secret.Data == nil { + secret.Data = map[string][]byte{} + } + secret.Data[controller.TokenKey] = []byte(token) + if err := s.Client.Patch(ctx, &secret, original); err != nil { + return nil, status.Errorf(codes.Internal, "failed to update credential secret: %s", err) + } + + log.FromContext(ctx).Info("token rotated", "exporter", exporter.Name, "namespace", exporter.Namespace) + + claims := &struct { + jwt.RegisteredClaims + }{} + parser := jwt.NewParser(jwt.WithoutClaimsValidation()) + if _, _, err := parser.ParseUnverified(token, claims); err != nil { + return nil, status.Errorf(codes.Internal, "failed to parse token claims: %s", err) + } + + var expiry *timestamppb.Timestamp + if claims.ExpiresAt != nil { + expiry = timestamppb.New(claims.ExpiresAt.Time) + } + + return &pb.RotateTokenResponse{ + Token: token, + Expiry: expiry, + }, nil +} + func (s *ControllerService) Unregister( ctx context.Context, req *pb.UnregisterRequest, diff --git a/controller/internal/service/controller_service_test.go b/controller/internal/service/controller_service_test.go index 2848d3cbd..5934af795 100644 --- a/controller/internal/service/controller_service_test.go +++ b/controller/internal/service/controller_service_test.go @@ -32,15 +32,18 @@ import ( jumpstarterdevv1alpha1 "github.com/jumpstarter-dev/jumpstarter/controller/api/v1alpha1" "github.com/jumpstarter-dev/jumpstarter/controller/internal/config" jlog "github.com/jumpstarter-dev/jumpstarter/controller/internal/log" + "github.com/jumpstarter-dev/jumpstarter/controller/internal/oidc" pb "github.com/jumpstarter-dev/jumpstarter/controller/internal/protocol/jumpstarter/v1" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" grpcpeer "google.golang.org/grpc/peer" "google.golang.org/grpc/status" + corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" k8sruntime "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" "k8s.io/apiserver/pkg/authentication/authenticator" "k8s.io/apiserver/pkg/authentication/user" "k8s.io/apiserver/pkg/authorization/authorizer" @@ -2383,3 +2386,116 @@ func TestStatusResponseIncludesLeaseContext(t *testing.T) { } }) } + +func TestControllerServiceRotateToken(t *testing.T) { + scheme := k8sruntime.NewScheme() + if err := jumpstarterdevv1alpha1.AddToScheme(scheme); err != nil { + t.Fatalf("failed to add jumpstarter scheme: %v", err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("failed to add corev1 scheme: %v", err) + } + + signer, err := oidc.NewSignerFromSeed([]byte{}, "https://example.com", "dummy") + if err != nil { + t.Fatalf("failed to create signer: %v", err) + } + + t.Run("successful rotation", func(t *testing.T) { + exporter := &jumpstarterdevv1alpha1.Exporter{ + ObjectMeta: metav1.ObjectMeta{Name: "test-exporter", Namespace: "default"}, + } + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-exporter-exporter", + Namespace: "default", + }, + Data: map[string][]byte{ + "token": []byte("old-token"), + }, + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(exporter, secret).Build() + + svc := &ControllerService{ + Client: fakeClient, + Authn: &passingAuthenticator{userName: "test-user"}, + Authz: passingAuthorizer{}, + Attr: &exporterAttributesGetter{namespace: "default", name: "test-exporter"}, + Signer: signer, + } + + resp, err := svc.RotateToken(context.Background(), &pb.RotateTokenRequest{}) + if err != nil { + t.Fatalf("RotateToken failed: %v", err) + } + if resp.Token == "" { + t.Fatal("expected non-empty token") + } + if resp.Token == "old-token" { + t.Fatal("expected token to be different from old-token") + } + if resp.Expiry == nil { + t.Fatal("expected expiry timestamp") + } + + var updatedSecret corev1.Secret + if err := fakeClient.Get(context.Background(), types.NamespacedName{ + Namespace: "default", + Name: "test-exporter-exporter", + }, &updatedSecret); err != nil { + t.Fatalf("failed to get updated secret: %v", err) + } + if string(updatedSecret.Data["token"]) != resp.Token { + t.Fatalf("expected secret token %q, got %q", resp.Token, string(updatedSecret.Data["token"])) + } + }) + + t.Run("missing signer returns failed precondition", func(t *testing.T) { + exporter := &jumpstarterdevv1alpha1.Exporter{ + ObjectMeta: metav1.ObjectMeta{Name: "test-exporter", Namespace: "default"}, + } + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "test-exporter-exporter", Namespace: "default"}, + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(exporter, secret).Build() + + svc := &ControllerService{ + Client: fakeClient, + Authn: &passingAuthenticator{userName: "test-user"}, + Authz: passingAuthorizer{}, + Attr: &exporterAttributesGetter{namespace: "default", name: "test-exporter"}, + Signer: nil, + } + + _, err := svc.RotateToken(context.Background(), &pb.RotateTokenRequest{}) + if err == nil { + t.Fatal("expected error with nil signer") + } + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("expected FailedPrecondition, got %v", err) + } + }) + + t.Run("missing secret returns internal error", func(t *testing.T) { + exporter := &jumpstarterdevv1alpha1.Exporter{ + ObjectMeta: metav1.ObjectMeta{Name: "test-exporter", Namespace: "default"}, + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(exporter).Build() + + svc := &ControllerService{ + Client: fakeClient, + Authn: &passingAuthenticator{userName: "test-user"}, + Authz: passingAuthorizer{}, + Attr: &exporterAttributesGetter{namespace: "default", name: "test-exporter"}, + Signer: signer, + } + + _, err := svc.RotateToken(context.Background(), &pb.RotateTokenRequest{}) + if err == nil { + t.Fatal("expected error with missing secret") + } + if status.Code(err) != codes.Internal { + t.Fatalf("expected Internal error, got %v", err) + } + }) +} diff --git a/protocol/proto/jumpstarter/v1/jumpstarter.proto b/protocol/proto/jumpstarter/v1/jumpstarter.proto index 364a1e105..21f7276ea 100644 --- a/protocol/proto/jumpstarter/v1/jumpstarter.proto +++ b/protocol/proto/jumpstarter/v1/jumpstarter.proto @@ -49,6 +49,9 @@ service ControllerService { // Returns an empty list when no optional services are deployed. // Older controllers return UNIMPLEMENTED; callers must treat that as an empty list. rpc GetServiceEndpoints(GetServiceEndpointsRequest) returns (GetServiceEndpointsResponse); + + // Rotate the authentication token for the exporter. + rpc RotateToken(RotateTokenRequest) returns (RotateTokenResponse); } // Registration request sent by an exporter to the controller. @@ -280,4 +283,13 @@ message TelemetryEndpoint { string endpoint = 1; // gRPC address (host:port). string certificate = 2; // Optional CA certificate PEM for TLS verification. string min_severity = 3; // Minimum log severity to forward (e.g. "info", "warning"). +} + +// Request to rotate the authentication token for an exporter. +message RotateTokenRequest {} + +// Response containing the newly rotated token. +message RotateTokenResponse { + string token = 1; // The new authentication token. + google.protobuf.Timestamp expiry = 2; // The expiration time of the new token. } \ No newline at end of file diff --git a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.py b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.py index 5ec31d09a..1d4cdd755 100644 --- a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.py +++ b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.py @@ -30,7 +30,7 @@ from . import common_pb2 as jumpstarter_dot_v1_dot_common__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n jumpstarter/v1/jumpstarter.proto\x12\x0ejumpstarter.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1bgoogle/protobuf/empty.proto\x1a\x1cgoogle/protobuf/struct.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1fjumpstarter/v1/kubernetes.proto\x1a\x1bjumpstarter/v1/common.proto\"\xd1\x01\n\x0fRegisterRequest\x12\x43\n\x06labels\x18\x01 \x03(\x0b\x32+.jumpstarter.v1.RegisterRequest.LabelsEntryR\x06labels\x12>\n\x07reports\x18\x02 \x03(\x0b\x32$.jumpstarter.v1.DriverInstanceReportR\x07reports\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xd2\x03\n\x14\x44riverInstanceReport\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12$\n\x0bparent_uuid\x18\x02 \x01(\tH\x00R\nparentUuid\x88\x01\x01\x12H\n\x06labels\x18\x03 \x03(\x0b\x32\x30.jumpstarter.v1.DriverInstanceReport.LabelsEntryR\x06labels\x12%\n\x0b\x64\x65scription\x18\x04 \x01(\tH\x01R\x0b\x64\x65scription\x88\x01\x01\x12m\n\x13methods_description\x18\x05 \x03(\x0b\x32<.jumpstarter.v1.DriverInstanceReport.MethodsDescriptionEntryR\x12methodsDescription\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x1a\x45\n\x17MethodsDescriptionEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\x0e\n\x0c_parent_uuidB\x0e\n\x0c_description\"&\n\x10RegisterResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\"+\n\x11UnregisterRequest\x12\x16\n\x06reason\x18\x02 \x01(\tR\x06reason\"\x14\n\x12UnregisterResponse\".\n\rListenRequest\x12\x1d\n\nlease_name\x18\x01 \x01(\tR\tleaseName\"\\\n\x0eListenResponse\x12\'\n\x0frouter_endpoint\x18\x01 \x01(\tR\x0erouterEndpoint\x12!\n\x0crouter_token\x18\x02 \x01(\tR\x0brouterToken\"\x0f\n\rStatusRequest\"\x94\x02\n\x0eStatusResponse\x12\x16\n\x06leased\x18\x01 \x01(\x08R\x06leased\x12\"\n\nlease_name\x18\x02 \x01(\tH\x00R\tleaseName\x88\x01\x01\x12$\n\x0b\x63lient_name\x18\x03 \x01(\tH\x01R\nclientName\x88\x01\x01\x12\x45\n\x07\x63ontext\x18\x04 \x03(\x0b\x32+.jumpstarter.v1.StatusResponse.ContextEntryR\x07\x63ontext\x1a:\n\x0c\x43ontextEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\r\n\x0b_lease_nameB\x0e\n\x0c_client_name\",\n\x0b\x44ialRequest\x12\x1d\n\nlease_name\x18\x01 \x01(\tR\tleaseName\"Z\n\x0c\x44ialResponse\x12\'\n\x0frouter_endpoint\x18\x01 \x01(\tR\x0erouterEndpoint\x12!\n\x0crouter_token\x18\x02 \x01(\tR\x0brouterToken\"\xb4\x01\n\x13ReportStatusRequest\x12\x36\n\x06status\x18\x01 \x01(\x0e\x32\x1e.jumpstarter.v1.ExporterStatusR\x06status\x12\x1d\n\x07message\x18\x02 \x01(\tH\x00R\x07message\x88\x01\x01\x12(\n\rrelease_lease\x18\x03 \x01(\x08H\x01R\x0creleaseLease\x88\x01\x01\x42\n\n\x08_messageB\x10\n\x0e_release_lease\"\x16\n\x14ReportStatusResponse\"\xcc\x02\n\x11GetReportResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x45\n\x06labels\x18\x02 \x03(\x0b\x32-.jumpstarter.v1.GetReportResponse.LabelsEntryR\x06labels\x12>\n\x07reports\x18\x03 \x03(\x0b\x32$.jumpstarter.v1.DriverInstanceReportR\x07reports\x12M\n\x15\x61lternative_endpoints\x18\x04 \x03(\x0b\x32\x18.jumpstarter.v1.EndpointR\x14\x61lternativeEndpoints\x12\x12\n\x04motd\x18\x05 \x01(\tR\x04motd\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xa5\x01\n\x08\x45ndpoint\x12\x1a\n\x08\x65ndpoint\x18\x01 \x01(\tR\x08\x65ndpoint\x12 \n\x0b\x63\x65rtificate\x18\x02 \x01(\tR\x0b\x63\x65rtificate\x12-\n\x12\x63lient_certificate\x18\x03 \x01(\tR\x11\x63lientCertificate\x12,\n\x12\x63lient_private_key\x18\x04 \x01(\tR\x10\x63lientPrivateKey\"k\n\x11\x44riverCallRequest\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x16\n\x06method\x18\x02 \x01(\tR\x06method\x12*\n\x04\x61rgs\x18\x03 \x03(\x0b\x32\x16.google.protobuf.ValueR\x04\x61rgs\"X\n\x12\x44riverCallResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12.\n\x06result\x18\x02 \x01(\x0b\x32\x16.google.protobuf.ValueR\x06result\"t\n\x1aStreamingDriverCallRequest\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x16\n\x06method\x18\x02 \x01(\tR\x06method\x12*\n\x04\x61rgs\x18\x03 \x03(\x0b\x32\x16.google.protobuf.ValueR\x04\x61rgs\"a\n\x1bStreamingDriverCallResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12.\n\x06result\x18\x02 \x01(\x0b\x32\x16.google.protobuf.ValueR\x06result\"\xff\x03\n\x11LogStreamResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x1a\n\x08severity\x18\x02 \x01(\tR\x08severity\x12\x18\n\x07message\x18\x03 \x01(\tR\x07message\x12\x36\n\x06source\x18\x04 \x01(\x0e\x32\x19.jumpstarter.v1.LogSourceH\x00R\x06source\x88\x01\x01\x12$\n\x0b\x64river_type\x18\x05 \x01(\tH\x01R\ndriverType\x88\x01\x01\x12!\n\toperation\x18\x06 \x01(\tH\x02R\toperation\x88\x01\x01\x12=\n\ttimestamp\x18\x07 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x03R\ttimestamp\x88\x01\x01\x12\x64\n\x11structured_fields\x18\x08 \x03(\x0b\x32\x37.jumpstarter.v1.LogStreamResponse.StructuredFieldsEntryR\x10structuredFields\x1a\x43\n\x15StructuredFieldsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\t\n\x07_sourceB\x0e\n\x0c_driver_typeB\x0c\n\n_operationB\x0c\n\n_timestamp\"\x0e\n\x0cResetRequest\"\x0f\n\rResetResponse\"%\n\x0fGetLeaseRequest\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\"\x93\x03\n\x10GetLeaseResponse\x12\x35\n\x08\x64uration\x18\x01 \x01(\x0b\x32\x19.google.protobuf.DurationR\x08\x64uration\x12\x39\n\x08selector\x18\x02 \x01(\x0b\x32\x1d.jumpstarter.v1.LabelSelectorR\x08selector\x12>\n\nbegin_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x00R\tbeginTime\x88\x01\x01\x12:\n\x08\x65nd_time\x18\x04 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x01R\x07\x65ndTime\x88\x01\x01\x12(\n\rexporter_uuid\x18\x05 \x01(\tH\x02R\x0c\x65xporterUuid\x88\x01\x01\x12\x39\n\nconditions\x18\x06 \x03(\x0b\x32\x19.jumpstarter.v1.ConditionR\nconditionsB\r\n\x0b_begin_timeB\x0b\n\t_end_timeB\x10\n\x0e_exporter_uuid\"\x87\x01\n\x13RequestLeaseRequest\x12\x35\n\x08\x64uration\x18\x01 \x01(\x0b\x32\x19.google.protobuf.DurationR\x08\x64uration\x12\x39\n\x08selector\x18\x02 \x01(\x0b\x32\x1d.jumpstarter.v1.LabelSelectorR\x08selector\"*\n\x14RequestLeaseResponse\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\")\n\x13ReleaseLeaseRequest\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\"\x16\n\x14ReleaseLeaseResponse\"\x13\n\x11ListLeasesRequest\"*\n\x12ListLeasesResponse\x12\x14\n\x05names\x18\x01 \x03(\tR\x05names\"\x12\n\x10GetStatusRequest\"\xff\x01\n\x11GetStatusResponse\x12\x36\n\x06status\x18\x01 \x01(\x0e\x32\x1e.jumpstarter.v1.ExporterStatusR\x06status\x12\x1d\n\x07message\x18\x02 \x01(\tH\x00R\x07message\x88\x01\x01\x12%\n\x0estatus_version\x18\x03 \x01(\x04R\rstatusVersion\x12L\n\x0fprevious_status\x18\x04 \x01(\x0e\x32\x1e.jumpstarter.v1.ExporterStatusH\x01R\x0epreviousStatus\x88\x01\x01\x42\n\n\x08_messageB\x12\n\x10_previous_status\"\x13\n\x11\x45ndSessionRequest\"Y\n\x12\x45ndSessionResponse\x12\x18\n\x07success\x18\x01 \x01(\x08R\x07success\x12\x1d\n\x07message\x18\x02 \x01(\tH\x00R\x07message\x88\x01\x01\x42\n\n\x08_message\"\x1c\n\x1aGetServiceEndpointsRequest\"q\n\x1bGetServiceEndpointsResponse\x12R\n\x13telemetry_endpoints\x18\x01 \x03(\x0b\x32!.jumpstarter.v1.TelemetryEndpointR\x12telemetryEndpoints\"t\n\x11TelemetryEndpoint\x12\x1a\n\x08\x65ndpoint\x18\x01 \x01(\tR\x08\x65ndpoint\x12 \n\x0b\x63\x65rtificate\x18\x02 \x01(\tR\x0b\x63\x65rtificate\x12!\n\x0cmin_severity\x18\x03 \x01(\tR\x0bminSeverity2\xb5\x07\n\x11\x43ontrollerService\x12M\n\x08Register\x12\x1f.jumpstarter.v1.RegisterRequest\x1a .jumpstarter.v1.RegisterResponse\x12S\n\nUnregister\x12!.jumpstarter.v1.UnregisterRequest\x1a\".jumpstarter.v1.UnregisterResponse\x12Y\n\x0cReportStatus\x12#.jumpstarter.v1.ReportStatusRequest\x1a$.jumpstarter.v1.ReportStatusResponse\x12I\n\x06Listen\x12\x1d.jumpstarter.v1.ListenRequest\x1a\x1e.jumpstarter.v1.ListenResponse0\x01\x12I\n\x06Status\x12\x1d.jumpstarter.v1.StatusRequest\x1a\x1e.jumpstarter.v1.StatusResponse0\x01\x12\x41\n\x04\x44ial\x12\x1b.jumpstarter.v1.DialRequest\x1a\x1c.jumpstarter.v1.DialResponse\x12M\n\x08GetLease\x12\x1f.jumpstarter.v1.GetLeaseRequest\x1a .jumpstarter.v1.GetLeaseResponse\x12Y\n\x0cRequestLease\x12#.jumpstarter.v1.RequestLeaseRequest\x1a$.jumpstarter.v1.RequestLeaseResponse\x12Y\n\x0cReleaseLease\x12#.jumpstarter.v1.ReleaseLeaseRequest\x1a$.jumpstarter.v1.ReleaseLeaseResponse\x12S\n\nListLeases\x12!.jumpstarter.v1.ListLeasesRequest\x1a\".jumpstarter.v1.ListLeasesResponse\x12n\n\x13GetServiceEndpoints\x12*.jumpstarter.v1.GetServiceEndpointsRequest\x1a+.jumpstarter.v1.GetServiceEndpointsResponse2\xd7\x04\n\x0f\x45xporterService\x12\x46\n\tGetReport\x12\x16.google.protobuf.Empty\x1a!.jumpstarter.v1.GetReportResponse\x12S\n\nDriverCall\x12!.jumpstarter.v1.DriverCallRequest\x1a\".jumpstarter.v1.DriverCallResponse\x12p\n\x13StreamingDriverCall\x12*.jumpstarter.v1.StreamingDriverCallRequest\x1a+.jumpstarter.v1.StreamingDriverCallResponse0\x01\x12H\n\tLogStream\x12\x16.google.protobuf.Empty\x1a!.jumpstarter.v1.LogStreamResponse0\x01\x12\x44\n\x05Reset\x12\x1c.jumpstarter.v1.ResetRequest\x1a\x1d.jumpstarter.v1.ResetResponse\x12P\n\tGetStatus\x12 .jumpstarter.v1.GetStatusRequest\x1a!.jumpstarter.v1.GetStatusResponse\x12S\n\nEndSession\x12!.jumpstarter.v1.EndSessionRequest\x1a\".jumpstarter.v1.EndSessionResponseB\x7f\n\x12\x63om.jumpstarter.v1B\x10JumpstarterProtoP\x01\xa2\x02\x03JXX\xaa\x02\x0eJumpstarter.V1\xca\x02\x0eJumpstarter\\V1\xe2\x02\x1aJumpstarter\\V1\\GPBMetadata\xea\x02\x0fJumpstarter::V1b\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n jumpstarter/v1/jumpstarter.proto\x12\x0ejumpstarter.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1bgoogle/protobuf/empty.proto\x1a\x1cgoogle/protobuf/struct.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x1fjumpstarter/v1/kubernetes.proto\x1a\x1bjumpstarter/v1/common.proto\"\xd1\x01\n\x0fRegisterRequest\x12\x43\n\x06labels\x18\x01 \x03(\x0b\x32+.jumpstarter.v1.RegisterRequest.LabelsEntryR\x06labels\x12>\n\x07reports\x18\x02 \x03(\x0b\x32$.jumpstarter.v1.DriverInstanceReportR\x07reports\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xd2\x03\n\x14\x44riverInstanceReport\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12$\n\x0bparent_uuid\x18\x02 \x01(\tH\x00R\nparentUuid\x88\x01\x01\x12H\n\x06labels\x18\x03 \x03(\x0b\x32\x30.jumpstarter.v1.DriverInstanceReport.LabelsEntryR\x06labels\x12%\n\x0b\x64\x65scription\x18\x04 \x01(\tH\x01R\x0b\x64\x65scription\x88\x01\x01\x12m\n\x13methods_description\x18\x05 \x03(\x0b\x32<.jumpstarter.v1.DriverInstanceReport.MethodsDescriptionEntryR\x12methodsDescription\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x1a\x45\n\x17MethodsDescriptionEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\x0e\n\x0c_parent_uuidB\x0e\n\x0c_description\"&\n\x10RegisterResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\"+\n\x11UnregisterRequest\x12\x16\n\x06reason\x18\x02 \x01(\tR\x06reason\"\x14\n\x12UnregisterResponse\".\n\rListenRequest\x12\x1d\n\nlease_name\x18\x01 \x01(\tR\tleaseName\"\\\n\x0eListenResponse\x12\'\n\x0frouter_endpoint\x18\x01 \x01(\tR\x0erouterEndpoint\x12!\n\x0crouter_token\x18\x02 \x01(\tR\x0brouterToken\"\x0f\n\rStatusRequest\"\x94\x02\n\x0eStatusResponse\x12\x16\n\x06leased\x18\x01 \x01(\x08R\x06leased\x12\"\n\nlease_name\x18\x02 \x01(\tH\x00R\tleaseName\x88\x01\x01\x12$\n\x0b\x63lient_name\x18\x03 \x01(\tH\x01R\nclientName\x88\x01\x01\x12\x45\n\x07\x63ontext\x18\x04 \x03(\x0b\x32+.jumpstarter.v1.StatusResponse.ContextEntryR\x07\x63ontext\x1a:\n\x0c\x43ontextEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\r\n\x0b_lease_nameB\x0e\n\x0c_client_name\",\n\x0b\x44ialRequest\x12\x1d\n\nlease_name\x18\x01 \x01(\tR\tleaseName\"Z\n\x0c\x44ialResponse\x12\'\n\x0frouter_endpoint\x18\x01 \x01(\tR\x0erouterEndpoint\x12!\n\x0crouter_token\x18\x02 \x01(\tR\x0brouterToken\"\xb4\x01\n\x13ReportStatusRequest\x12\x36\n\x06status\x18\x01 \x01(\x0e\x32\x1e.jumpstarter.v1.ExporterStatusR\x06status\x12\x1d\n\x07message\x18\x02 \x01(\tH\x00R\x07message\x88\x01\x01\x12(\n\rrelease_lease\x18\x03 \x01(\x08H\x01R\x0creleaseLease\x88\x01\x01\x42\n\n\x08_messageB\x10\n\x0e_release_lease\"\x16\n\x14ReportStatusResponse\"\xcc\x02\n\x11GetReportResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x45\n\x06labels\x18\x02 \x03(\x0b\x32-.jumpstarter.v1.GetReportResponse.LabelsEntryR\x06labels\x12>\n\x07reports\x18\x03 \x03(\x0b\x32$.jumpstarter.v1.DriverInstanceReportR\x07reports\x12M\n\x15\x61lternative_endpoints\x18\x04 \x03(\x0b\x32\x18.jumpstarter.v1.EndpointR\x14\x61lternativeEndpoints\x12\x12\n\x04motd\x18\x05 \x01(\tR\x04motd\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xa5\x01\n\x08\x45ndpoint\x12\x1a\n\x08\x65ndpoint\x18\x01 \x01(\tR\x08\x65ndpoint\x12 \n\x0b\x63\x65rtificate\x18\x02 \x01(\tR\x0b\x63\x65rtificate\x12-\n\x12\x63lient_certificate\x18\x03 \x01(\tR\x11\x63lientCertificate\x12,\n\x12\x63lient_private_key\x18\x04 \x01(\tR\x10\x63lientPrivateKey\"k\n\x11\x44riverCallRequest\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x16\n\x06method\x18\x02 \x01(\tR\x06method\x12*\n\x04\x61rgs\x18\x03 \x03(\x0b\x32\x16.google.protobuf.ValueR\x04\x61rgs\"X\n\x12\x44riverCallResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12.\n\x06result\x18\x02 \x01(\x0b\x32\x16.google.protobuf.ValueR\x06result\"t\n\x1aStreamingDriverCallRequest\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x16\n\x06method\x18\x02 \x01(\tR\x06method\x12*\n\x04\x61rgs\x18\x03 \x03(\x0b\x32\x16.google.protobuf.ValueR\x04\x61rgs\"a\n\x1bStreamingDriverCallResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12.\n\x06result\x18\x02 \x01(\x0b\x32\x16.google.protobuf.ValueR\x06result\"\xff\x03\n\x11LogStreamResponse\x12\x12\n\x04uuid\x18\x01 \x01(\tR\x04uuid\x12\x1a\n\x08severity\x18\x02 \x01(\tR\x08severity\x12\x18\n\x07message\x18\x03 \x01(\tR\x07message\x12\x36\n\x06source\x18\x04 \x01(\x0e\x32\x19.jumpstarter.v1.LogSourceH\x00R\x06source\x88\x01\x01\x12$\n\x0b\x64river_type\x18\x05 \x01(\tH\x01R\ndriverType\x88\x01\x01\x12!\n\toperation\x18\x06 \x01(\tH\x02R\toperation\x88\x01\x01\x12=\n\ttimestamp\x18\x07 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x03R\ttimestamp\x88\x01\x01\x12\x64\n\x11structured_fields\x18\x08 \x03(\x0b\x32\x37.jumpstarter.v1.LogStreamResponse.StructuredFieldsEntryR\x10structuredFields\x1a\x43\n\x15StructuredFieldsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\t\n\x07_sourceB\x0e\n\x0c_driver_typeB\x0c\n\n_operationB\x0c\n\n_timestamp\"\x0e\n\x0cResetRequest\"\x0f\n\rResetResponse\"%\n\x0fGetLeaseRequest\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\"\x93\x03\n\x10GetLeaseResponse\x12\x35\n\x08\x64uration\x18\x01 \x01(\x0b\x32\x19.google.protobuf.DurationR\x08\x64uration\x12\x39\n\x08selector\x18\x02 \x01(\x0b\x32\x1d.jumpstarter.v1.LabelSelectorR\x08selector\x12>\n\nbegin_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x00R\tbeginTime\x88\x01\x01\x12:\n\x08\x65nd_time\x18\x04 \x01(\x0b\x32\x1a.google.protobuf.TimestampH\x01R\x07\x65ndTime\x88\x01\x01\x12(\n\rexporter_uuid\x18\x05 \x01(\tH\x02R\x0c\x65xporterUuid\x88\x01\x01\x12\x39\n\nconditions\x18\x06 \x03(\x0b\x32\x19.jumpstarter.v1.ConditionR\nconditionsB\r\n\x0b_begin_timeB\x0b\n\t_end_timeB\x10\n\x0e_exporter_uuid\"\x87\x01\n\x13RequestLeaseRequest\x12\x35\n\x08\x64uration\x18\x01 \x01(\x0b\x32\x19.google.protobuf.DurationR\x08\x64uration\x12\x39\n\x08selector\x18\x02 \x01(\x0b\x32\x1d.jumpstarter.v1.LabelSelectorR\x08selector\"*\n\x14RequestLeaseResponse\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\")\n\x13ReleaseLeaseRequest\x12\x12\n\x04name\x18\x01 \x01(\tR\x04name\"\x16\n\x14ReleaseLeaseResponse\"\x13\n\x11ListLeasesRequest\"*\n\x12ListLeasesResponse\x12\x14\n\x05names\x18\x01 \x03(\tR\x05names\"\x12\n\x10GetStatusRequest\"\xff\x01\n\x11GetStatusResponse\x12\x36\n\x06status\x18\x01 \x01(\x0e\x32\x1e.jumpstarter.v1.ExporterStatusR\x06status\x12\x1d\n\x07message\x18\x02 \x01(\tH\x00R\x07message\x88\x01\x01\x12%\n\x0estatus_version\x18\x03 \x01(\x04R\rstatusVersion\x12L\n\x0fprevious_status\x18\x04 \x01(\x0e\x32\x1e.jumpstarter.v1.ExporterStatusH\x01R\x0epreviousStatus\x88\x01\x01\x42\n\n\x08_messageB\x12\n\x10_previous_status\"\x13\n\x11\x45ndSessionRequest\"Y\n\x12\x45ndSessionResponse\x12\x18\n\x07success\x18\x01 \x01(\x08R\x07success\x12\x1d\n\x07message\x18\x02 \x01(\tH\x00R\x07message\x88\x01\x01\x42\n\n\x08_message\"\x1c\n\x1aGetServiceEndpointsRequest\"q\n\x1bGetServiceEndpointsResponse\x12R\n\x13telemetry_endpoints\x18\x01 \x03(\x0b\x32!.jumpstarter.v1.TelemetryEndpointR\x12telemetryEndpoints\"t\n\x11TelemetryEndpoint\x12\x1a\n\x08\x65ndpoint\x18\x01 \x01(\tR\x08\x65ndpoint\x12 \n\x0b\x63\x65rtificate\x18\x02 \x01(\tR\x0b\x63\x65rtificate\x12!\n\x0cmin_severity\x18\x03 \x01(\tR\x0bminSeverity\"\x14\n\x12RotateTokenRequest\"_\n\x13RotateTokenResponse\x12\x14\n\x05token\x18\x01 \x01(\tR\x05token\x12\x32\n\x06\x65xpiry\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.TimestampR\x06\x65xpiry2\x8d\x08\n\x11\x43ontrollerService\x12M\n\x08Register\x12\x1f.jumpstarter.v1.RegisterRequest\x1a .jumpstarter.v1.RegisterResponse\x12S\n\nUnregister\x12!.jumpstarter.v1.UnregisterRequest\x1a\".jumpstarter.v1.UnregisterResponse\x12Y\n\x0cReportStatus\x12#.jumpstarter.v1.ReportStatusRequest\x1a$.jumpstarter.v1.ReportStatusResponse\x12I\n\x06Listen\x12\x1d.jumpstarter.v1.ListenRequest\x1a\x1e.jumpstarter.v1.ListenResponse0\x01\x12I\n\x06Status\x12\x1d.jumpstarter.v1.StatusRequest\x1a\x1e.jumpstarter.v1.StatusResponse0\x01\x12\x41\n\x04\x44ial\x12\x1b.jumpstarter.v1.DialRequest\x1a\x1c.jumpstarter.v1.DialResponse\x12M\n\x08GetLease\x12\x1f.jumpstarter.v1.GetLeaseRequest\x1a .jumpstarter.v1.GetLeaseResponse\x12Y\n\x0cRequestLease\x12#.jumpstarter.v1.RequestLeaseRequest\x1a$.jumpstarter.v1.RequestLeaseResponse\x12Y\n\x0cReleaseLease\x12#.jumpstarter.v1.ReleaseLeaseRequest\x1a$.jumpstarter.v1.ReleaseLeaseResponse\x12S\n\nListLeases\x12!.jumpstarter.v1.ListLeasesRequest\x1a\".jumpstarter.v1.ListLeasesResponse\x12n\n\x13GetServiceEndpoints\x12*.jumpstarter.v1.GetServiceEndpointsRequest\x1a+.jumpstarter.v1.GetServiceEndpointsResponse\x12V\n\x0bRotateToken\x12\".jumpstarter.v1.RotateTokenRequest\x1a#.jumpstarter.v1.RotateTokenResponse2\xd7\x04\n\x0f\x45xporterService\x12\x46\n\tGetReport\x12\x16.google.protobuf.Empty\x1a!.jumpstarter.v1.GetReportResponse\x12S\n\nDriverCall\x12!.jumpstarter.v1.DriverCallRequest\x1a\".jumpstarter.v1.DriverCallResponse\x12p\n\x13StreamingDriverCall\x12*.jumpstarter.v1.StreamingDriverCallRequest\x1a+.jumpstarter.v1.StreamingDriverCallResponse0\x01\x12H\n\tLogStream\x12\x16.google.protobuf.Empty\x1a!.jumpstarter.v1.LogStreamResponse0\x01\x12\x44\n\x05Reset\x12\x1c.jumpstarter.v1.ResetRequest\x1a\x1d.jumpstarter.v1.ResetResponse\x12P\n\tGetStatus\x12 .jumpstarter.v1.GetStatusRequest\x1a!.jumpstarter.v1.GetStatusResponse\x12S\n\nEndSession\x12!.jumpstarter.v1.EndSessionRequest\x1a\".jumpstarter.v1.EndSessionResponseB\x7f\n\x12\x63om.jumpstarter.v1B\x10JumpstarterProtoP\x01\xa2\x02\x03JXX\xaa\x02\x0eJumpstarter.V1\xca\x02\x0eJumpstarter\\V1\xe2\x02\x1aJumpstarter\\V1\\GPBMetadata\xea\x02\x0fJumpstarter::V1b\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -136,8 +136,12 @@ _globals['_GETSERVICEENDPOINTSRESPONSE']._serialized_end=4567 _globals['_TELEMETRYENDPOINT']._serialized_start=4569 _globals['_TELEMETRYENDPOINT']._serialized_end=4685 - _globals['_CONTROLLERSERVICE']._serialized_start=4688 - _globals['_CONTROLLERSERVICE']._serialized_end=5637 - _globals['_EXPORTERSERVICE']._serialized_start=5640 - _globals['_EXPORTERSERVICE']._serialized_end=6239 + _globals['_ROTATETOKENREQUEST']._serialized_start=4687 + _globals['_ROTATETOKENREQUEST']._serialized_end=4707 + _globals['_ROTATETOKENRESPONSE']._serialized_start=4709 + _globals['_ROTATETOKENRESPONSE']._serialized_end=4804 + _globals['_CONTROLLERSERVICE']._serialized_start=4807 + _globals['_CONTROLLERSERVICE']._serialized_end=5844 + _globals['_EXPORTERSERVICE']._serialized_start=5847 + _globals['_EXPORTERSERVICE']._serialized_end=6446 # @@protoc_insertion_point(module_scope) diff --git a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pyi b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pyi index 4dcf00b24..a10177a93 100644 --- a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pyi +++ b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2.pyi @@ -987,3 +987,40 @@ class TelemetryEndpoint(google.protobuf.message.Message): def ClearField(self, field_name: typing.Literal["certificate", b"certificate", "endpoint", b"endpoint", "min_severity", b"min_severity"]) -> None: ... Global___TelemetryEndpoint: typing_extensions.TypeAlias = TelemetryEndpoint + +@typing.final +class RotateTokenRequest(google.protobuf.message.Message): + """Request to rotate the authentication token for an exporter.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +Global___RotateTokenRequest: typing_extensions.TypeAlias = RotateTokenRequest + +@typing.final +class RotateTokenResponse(google.protobuf.message.Message): + """Response containing the newly rotated token.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + TOKEN_FIELD_NUMBER: builtins.int + EXPIRY_FIELD_NUMBER: builtins.int + token: builtins.str + """The new authentication token.""" + @property + def expiry(self) -> google.protobuf.timestamp_pb2.Timestamp: + """The expiration time of the new token.""" + + def __init__( + self, + *, + token: builtins.str = ..., + expiry: google.protobuf.timestamp_pb2.Timestamp | None = ..., + ) -> None: ... + def HasField(self, field_name: typing.Literal["expiry", b"expiry"]) -> builtins.bool: ... + def ClearField(self, field_name: typing.Literal["expiry", b"expiry", "token", b"token"]) -> None: ... + +Global___RotateTokenResponse: typing_extensions.TypeAlias = RotateTokenResponse diff --git a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.py b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.py index d7bb170af..182018134 100644 --- a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.py +++ b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.py @@ -71,6 +71,11 @@ def __init__(self, channel): request_serializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.GetServiceEndpointsRequest.SerializeToString, response_deserializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.GetServiceEndpointsResponse.FromString, _registered_method=True) + self.RotateToken = channel.unary_unary( + '/jumpstarter.v1.ControllerService/RotateToken', + request_serializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.RotateTokenRequest.SerializeToString, + response_deserializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.RotateTokenResponse.FromString, + _registered_method=True) class ControllerServiceServicer: @@ -161,6 +166,13 @@ def GetServiceEndpoints(self, request, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def RotateToken(self, request, context): + """Rotate the authentication token for the exporter. + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def add_ControllerServiceServicer_to_server(servicer, server): rpc_method_handlers = { @@ -219,6 +231,11 @@ def add_ControllerServiceServicer_to_server(servicer, server): request_deserializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.GetServiceEndpointsRequest.FromString, response_serializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.GetServiceEndpointsResponse.SerializeToString, ), + 'RotateToken': grpc.unary_unary_rpc_method_handler( + servicer.RotateToken, + request_deserializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.RotateTokenRequest.FromString, + response_serializer=jumpstarter_dot_v1_dot_jumpstarter__pb2.RotateTokenResponse.SerializeToString, + ), } generic_handler = grpc.method_handlers_generic_handler( 'jumpstarter.v1.ControllerService', rpc_method_handlers) @@ -528,6 +545,33 @@ def GetServiceEndpoints(request, metadata, _registered_method=True) + @staticmethod + def RotateToken(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/jumpstarter.v1.ControllerService/RotateToken', + jumpstarter_dot_v1_dot_jumpstarter__pb2.RotateTokenRequest.SerializeToString, + jumpstarter_dot_v1_dot_jumpstarter__pb2.RotateTokenResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + class ExporterServiceStub: """A service an exporter can share locally to be used without a server. diff --git a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pyi b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pyi index 1e4e79be5..ffcd842ca 100644 --- a/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pyi +++ b/python/packages/jumpstarter-protocol/jumpstarter_protocol/jumpstarter/v1/jumpstarter_pb2_grpc.pyi @@ -202,7 +202,23 @@ _ControllerServiceGetServiceEndpointsType = typing_extensions.TypeVar( ], ) -class ControllerServiceStub(typing.Generic[_ControllerServiceRegisterType, _ControllerServiceUnregisterType, _ControllerServiceReportStatusType, _ControllerServiceListenType, _ControllerServiceStatusType, _ControllerServiceDialType, _ControllerServiceGetLeaseType, _ControllerServiceRequestLeaseType, _ControllerServiceReleaseLeaseType, _ControllerServiceListLeasesType, _ControllerServiceGetServiceEndpointsType]): +_ControllerServiceRotateTokenType = typing_extensions.TypeVar( + '_ControllerServiceRotateTokenType', + grpc.UnaryUnaryMultiCallable[ + jumpstarter.v1.jumpstarter_pb2.RotateTokenRequest, + jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse, + ], + grpc.aio.UnaryUnaryMultiCallable[ + jumpstarter.v1.jumpstarter_pb2.RotateTokenRequest, + jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse, + ], + default=grpc.UnaryUnaryMultiCallable[ + jumpstarter.v1.jumpstarter_pb2.RotateTokenRequest, + jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse, + ], +) + +class ControllerServiceStub(typing.Generic[_ControllerServiceRegisterType, _ControllerServiceUnregisterType, _ControllerServiceReportStatusType, _ControllerServiceListenType, _ControllerServiceStatusType, _ControllerServiceDialType, _ControllerServiceGetLeaseType, _ControllerServiceRequestLeaseType, _ControllerServiceReleaseLeaseType, _ControllerServiceListLeasesType, _ControllerServiceGetServiceEndpointsType, _ControllerServiceRotateTokenType]): """A service where an exporter can connect to make itself available.""" @typing.overload @@ -251,6 +267,10 @@ class ControllerServiceStub(typing.Generic[_ControllerServiceRegisterType, _Cont jumpstarter.v1.jumpstarter_pb2.GetServiceEndpointsRequest, jumpstarter.v1.jumpstarter_pb2.GetServiceEndpointsResponse, ], + grpc.UnaryUnaryMultiCallable[ + jumpstarter.v1.jumpstarter_pb2.RotateTokenRequest, + jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse, + ], ], channel: grpc.Channel) -> None: ... @typing.overload @@ -299,6 +319,10 @@ class ControllerServiceStub(typing.Generic[_ControllerServiceRegisterType, _Cont jumpstarter.v1.jumpstarter_pb2.GetServiceEndpointsRequest, jumpstarter.v1.jumpstarter_pb2.GetServiceEndpointsResponse, ], + grpc.aio.UnaryUnaryMultiCallable[ + jumpstarter.v1.jumpstarter_pb2.RotateTokenRequest, + jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse, + ], ], channel: grpc.aio.Channel) -> None: ... Register: _ControllerServiceRegisterType @@ -345,6 +369,9 @@ class ControllerServiceStub(typing.Generic[_ControllerServiceRegisterType, _Cont Older controllers return UNIMPLEMENTED; callers must treat that as an empty list. """ + RotateToken: _ControllerServiceRotateTokenType + """Rotate the authentication token for the exporter.""" + ControllerServiceAsyncStub: typing_extensions.TypeAlias = ControllerServiceStub[ grpc.aio.UnaryUnaryMultiCallable[ jumpstarter.v1.jumpstarter_pb2.RegisterRequest, @@ -390,6 +417,10 @@ ControllerServiceAsyncStub: typing_extensions.TypeAlias = ControllerServiceStub[ jumpstarter.v1.jumpstarter_pb2.GetServiceEndpointsRequest, jumpstarter.v1.jumpstarter_pb2.GetServiceEndpointsResponse, ], + grpc.aio.UnaryUnaryMultiCallable[ + jumpstarter.v1.jumpstarter_pb2.RotateTokenRequest, + jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse, + ], ] class ControllerServiceServicer(metaclass=abc.ABCMeta): @@ -494,6 +525,14 @@ class ControllerServiceServicer(metaclass=abc.ABCMeta): Older controllers return UNIMPLEMENTED; callers must treat that as an empty list. """ + @abc.abstractmethod + def RotateToken( + self, + request: jumpstarter.v1.jumpstarter_pb2.RotateTokenRequest, + context: _ServicerContext, + ) -> typing.Union[jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse, collections.abc.Awaitable[jumpstarter.v1.jumpstarter_pb2.RotateTokenResponse]]: + """Rotate the authentication token for the exporter.""" + def add_ControllerServiceServicer_to_server(servicer: ControllerServiceServicer, server: typing.Union[grpc.Server, grpc.aio.Server]) -> None: ... _ExporterServiceGetReportType = typing_extensions.TypeVar( diff --git a/python/packages/jumpstarter/jumpstarter/config/exporter.py b/python/packages/jumpstarter/jumpstarter/config/exporter.py index 6d57c9434..5349c3824 100644 --- a/python/packages/jumpstarter/jumpstarter/config/exporter.py +++ b/python/packages/jumpstarter/jumpstarter/config/exporter.py @@ -1,6 +1,7 @@ from __future__ import annotations import errno +import logging import os import tempfile from contextlib import asynccontextmanager, contextmanager, suppress @@ -22,6 +23,8 @@ if TYPE_CHECKING: from jumpstarter.driver import Driver +logger = logging.getLogger(__name__) + class HookInstanceConfigV1Alpha1(BaseModel): """Configuration for a specific lifecycle hook.""" @@ -288,7 +291,9 @@ def save(cls, config: Self, path: Optional[str] = None) -> Path: with os.fdopen(temp_fd, "w") as f: yaml.safe_dump( config.model_dump( - mode="json", by_alias=True, exclude=cls._EXCLUDE_FROM_YAML, + mode="json", + by_alias=True, + exclude=cls._EXCLUDE_FROM_YAML, ), f, sort_keys=False, @@ -378,6 +383,15 @@ async def dummy_channel_factory() -> grpc.aio.Channel: config=self.hooks, ) + async def on_token_rotated(new_token: str): + self.token = new_token + if self.path is not None: + try: + self.save(self, path=str(self.path)) + logger.info("Saved rotated exporter token to %s", self.path) + except OSError as e: + logger.warning("Could not save rotated token to %s: %s", self.path, e) + exporter = None entered = False try: @@ -395,6 +409,7 @@ async def dummy_channel_factory() -> grpc.aio.Channel: hook_executor=hook_executor, motd=self.motd, exit_on_lease_end=self.exit_on_lease_end, + on_token_rotated=on_token_rotated, ) # Initialize the exporter (registration, etc.) await exporter.__aenter__() @@ -406,6 +421,11 @@ async def dummy_channel_factory() -> grpc.aio.Channel: with CancelScope(shield=True): await exporter.__aexit__(None, None, None) + async def rotate_token(self) -> str: + """Rotate the exporter authentication token with the controller.""" + async with self.create_exporter() as exporter: + return await exporter.rotate_token() + async def serve(self): async with self.create_exporter() as exporter: await exporter.serve() diff --git a/python/packages/jumpstarter/jumpstarter/exporter/exporter.py b/python/packages/jumpstarter/jumpstarter/exporter/exporter.py index af7fe210e..07ddab006 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/exporter.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/exporter.py @@ -1,3 +1,4 @@ +import inspect import logging import math import os @@ -36,6 +37,11 @@ from jumpstarter.exporter.lease_context import LeaseContext from jumpstarter.exporter.session import Session from jumpstarter.exporter.telemetry import TelemetryLogHandler +from jumpstarter.exporter.token_refresh import ( + DEFAULT_TOKEN_REFRESH_FRACTION, + DEFAULT_TOKEN_REFRESH_LEAD_TIME, + calculate_token_refresh_sleep, +) from jumpstarter.logging import clear_log_context, set_log_context if TYPE_CHECKING: @@ -60,12 +66,14 @@ _LEASE_FINISHED_WATCHDOG = 30.0 # Status codes indicating old controller without exporter auth on ReleaseLease -_RELEASE_LEASE_UNSUPPORTED_CODES = frozenset({ - grpc.StatusCode.PERMISSION_DENIED, - grpc.StatusCode.INVALID_ARGUMENT, - grpc.StatusCode.UNAUTHENTICATED, - grpc.StatusCode.UNIMPLEMENTED, -}) +_RELEASE_LEASE_UNSUPPORTED_CODES = frozenset( + { + grpc.StatusCode.PERMISSION_DENIED, + grpc.StatusCode.INVALID_ARGUMENT, + grpc.StatusCode.UNAUTHENTICATED, + grpc.StatusCode.UNIMPLEMENTED, + } +) _SEVERITY_MAP = { "debug": logging.DEBUG, @@ -83,6 +91,7 @@ def _severity_to_level(severity: str) -> int: if level is None: if key: import structlog + structlog.get_logger(__name__).warning( "Unrecognized min_severity value, defaulting to info", value=severity, @@ -260,6 +269,18 @@ class Exporter(AsyncContextManagerMixin, Metadata): the exporter's JWT as an Authorization header. """ + on_token_rotated: Callable[[str], Awaitable[None] | None] | None = field(default=None) + """Optional callback invoked when the authentication token is rotated. + + Passed the new token string. Can be async or sync. + """ + + token_refresh_lead_time: float = DEFAULT_TOKEN_REFRESH_LEAD_TIME + """Maximum lead time before expiry to attempt token refresh (default: 7 days).""" + + token_refresh_fraction: float = DEFAULT_TOKEN_REFRESH_FRACTION + """Fraction of total token lifetime at which to trigger refresh (default: 0.2).""" + # Internal State Fields _registered: bool = field(init=False, default=False) @@ -585,9 +606,7 @@ async def _setup_telemetry(self) -> None: logger.info("Connecting to telemetry service at %s (min_severity=%s)", ep.endpoint, ep.min_severity) grpc_insecure = ( - self.tls.insecure - or os.getenv(JMP_GRPC_INSECURE) == "1" - or os.getenv(JUMPSTARTER_GRPC_INSECURE) == "1" + self.tls.insecure or os.getenv(JMP_GRPC_INSECURE) == "1" or os.getenv(JUMPSTARTER_GRPC_INSECURE) == "1" ) if ep.certificate: @@ -601,9 +620,7 @@ async def _setup_telemetry(self) -> None: self._telemetry_channel = grpc.aio.insecure_channel(ep.endpoint) else: # Production: TLS with system CA pool. - self._telemetry_channel = grpc.aio.secure_channel( - ep.endpoint, grpc.ssl_channel_credentials() - ) + self._telemetry_channel = grpc.aio.secure_channel(ep.endpoint, grpc.ssl_channel_credentials()) stub = telemetry_pb2_grpc.TelemetryServiceStub(self._telemetry_channel) handler = TelemetryLogHandler(stub, namespace=getattr(self, "namespace", "") or "", token=self.token) handler.setLevel(_severity_to_level(ep.min_severity)) @@ -611,6 +628,95 @@ async def _setup_telemetry(self) -> None: self._telemetry_handler = handler logger.info("Telemetry log handler attached") + async def rotate_token(self) -> str: + """Request a newly signed authentication token from the controller. + + Updates self.token, updates the telemetry handler if attached, and invokes + the on_token_rotated callback if configured. + + Returns: + The newly rotated token string. + """ + async with self._controller_stub() as controller: + response: jumpstarter_pb2.RotateTokenResponse = await controller.RotateToken( + jumpstarter_pb2.RotateTokenRequest(), + timeout=_RPC_TIMEOUT, + ) + + new_token = response.token + self.token = new_token + + if self._telemetry_handler is not None: + self._telemetry_handler.token = new_token + + if self.on_token_rotated is not None: + try: + res = self.on_token_rotated(new_token) + if inspect.isawaitable(res): + await res + except Exception as e: + logger.error("Error in on_token_rotated callback: %s", e, exc_info=True) + + return new_token + + async def _token_refresh_loop(self) -> None: + """Monitor token expiration and request a new token when it is about to expire.""" + if self._standalone or not self.token: + return + + while True: + sleep_duration = calculate_token_refresh_sleep( + self.token, + lead_time=self.token_refresh_lead_time, + fraction=self.token_refresh_fraction, + ) + if sleep_duration is None: + logger.debug("Token does not have an expiry claim; automatic refresh disabled") + return + + if sleep_duration > 0: + logger.debug("Token refresh scheduled in %.1f seconds", sleep_duration) + await sleep(sleep_duration) + + # Token is about to expire (or already expired) - request a new token with retries + attempt = 0 + while True: + try: + logger.info("Token is about to expire; requesting new token from controller") + await self.rotate_token() + logger.info("Successfully refreshed exporter authentication token") + break + except grpc.aio.AioRpcError as e: + if e.code() == grpc.StatusCode.UNIMPLEMENTED: + logger.warning( + "Controller does not support token rotation (UNIMPLEMENTED); disabling automatic refresh" + ) + return + elif e.code() in (grpc.StatusCode.UNAUTHENTICATED, grpc.StatusCode.PERMISSION_DENIED): + logger.error("Failed to rotate token (%s): %s", e.code(), e.details()) + return + else: + attempt += 1 + backoff = min(_RPC_BACKOFF_BASE * (2 ** min(attempt - 1, 5)), _RPC_BACKOFF_CAP) + logger.warning( + "Transient failure rotating token (%s: %s); retrying in %.1fs (attempt %d)", + e.code(), + e.details(), + backoff, + attempt, + ) + await sleep(backoff) + except Exception as e: + attempt += 1 + backoff = min(_RPC_BACKOFF_BASE * (2 ** min(attempt - 1, 5)), _RPC_BACKOFF_CAP) + logger.warning( + "Unexpected failure rotating token (%s); retrying in %.1fs (attempt %d)", + e, + backoff, + attempt, + ) + await sleep(backoff) + async def _retry_rpc( self, rpc_call: Callable, @@ -1005,24 +1111,21 @@ async def _cleanup_after_lease(self, lease_scope: LeaseContext) -> None: # Use the configured hook timeout (+ margin) when available so we # never interrupt a legitimately-running beforeLease hook. safety_timeout = 15 # generous default for no-hook / unknown cases - if ( - self.hook_executor - and self.hook_executor.config.before_lease - ): + if self.hook_executor and self.hook_executor.config.before_lease: safety_timeout = self.hook_executor.config.before_lease.timeout + 30 with move_on_after(safety_timeout) as timeout_scope: await lease_scope.before_lease_hook.wait() if timeout_scope.cancelled_caught: - logger.warning( - "Timed out waiting for before_lease_hook; forcing it set to avoid deadlock" - ) + logger.warning("Timed out waiting for before_lease_hook; forcing it set to avoid deadlock") lease_scope.before_lease_hook.set() if not lease_scope.after_lease_hook_started.is_set(): lease_scope.after_lease_hook_started.set() - if (self.hook_executor - and (lease_scope.has_client() or self._standalone) - and not lease_scope.skip_after_lease_hook): + if ( + self.hook_executor + and (lease_scope.has_client() or self._standalone) + and not lease_scope.skip_after_lease_hook + ): logger.info("Running afterLease hook on session close") await self.hook_executor.run_after_lease_hook( lease_scope, @@ -1037,10 +1140,7 @@ async def _cleanup_after_lease(self, lease_scope: LeaseContext) -> None: else: logger.info("Skipping afterLease hook: beforeLease hook failed") if not self._stop_requested: - logger.debug( - "No afterLease hook or no client on session close," - " transitioning to AVAILABLE" - ) + logger.debug("No afterLease hook or no client on session close, transitioning to AVAILABLE") await self._report_status(ExporterStatus.AVAILABLE, "Available for new lease") else: logger.debug("Exporter is shutting down, skipping AVAILABLE status report") @@ -1304,6 +1404,7 @@ async def _run_control_plane( self._status_stream_factory(), status_tx, ) + tg.start_soon(self._token_refresh_loop) # One loop, one writer of _lease_context. Status ticks come from the # controller RPC; LeaseFinished comes from handle_lease when it has # torn a lease down. Both are handled here, sequentially. @@ -1559,7 +1660,9 @@ async def serve_standalone_tcp( lease_scope.hook_socket_path = hook_path_str async with session.serve_tcp_and_unix_async( - host, port, hook_path_str, + host, + port, + hook_path_str, tls_credentials=tls_credentials, interceptors=interceptors, ): diff --git a/python/packages/jumpstarter/jumpstarter/exporter/exporter_token_refresh_test.py b/python/packages/jumpstarter/jumpstarter/exporter/exporter_token_refresh_test.py new file mode 100644 index 000000000..d940d5a0f --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/exporter/exporter_token_refresh_test.py @@ -0,0 +1,245 @@ +import base64 +import json +import time +from unittest.mock import AsyncMock, MagicMock, patch + +import anyio +import grpc +import pytest +from jumpstarter_protocol import jumpstarter_pb2 + +from jumpstarter.config.common import ObjectMeta +from jumpstarter.config.exporter import ExporterConfigV1Alpha1 +from jumpstarter.config.tls import TLSConfigV1Alpha1 +from jumpstarter.exporter import Exporter + + +def _make_jwt(payload: dict) -> str: + header = {"alg": "ES256", "typ": "JWT"} + h_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).rstrip(b"=").decode() + p_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=").decode() + return f"{h_b64}.{p_b64}.dummy_signature" + + +class FakeRpcError(grpc.aio.AioRpcError): + def __init__(self, code, details="error"): + self._code = code + self._details = details + + def code(self): + return self._code + + def details(self): + return self._details + + +@pytest.mark.anyio +async def test_exporter_rotate_token_success(): + mock_channel = MagicMock(spec=grpc.aio.Channel) + mock_channel.close = AsyncMock() + + mock_ctrl = AsyncMock() + mock_ctrl.RotateToken.return_value = jumpstarter_pb2.RotateTokenResponse(token="rotated-token-value") + + async def channel_factory(): + return mock_channel + + on_rotated = AsyncMock() + mock_telemetry_handler = MagicMock() + mock_telemetry_handler.token = "old-token" + + exporter = Exporter( + exporter_name="test-exporter", + token="old-token", + channel_factory=channel_factory, + device_factory=lambda: MagicMock(), + on_token_rotated=on_rotated, + ) + exporter._telemetry_handler = mock_telemetry_handler + + with patch("jumpstarter.exporter.exporter.jumpstarter_pb2_grpc.ControllerServiceStub", return_value=mock_ctrl): + token = await exporter.rotate_token() + + assert token == "rotated-token-value" + assert exporter.token == "rotated-token-value" + assert mock_telemetry_handler.token == "rotated-token-value" + on_rotated.assert_awaited_once_with("rotated-token-value") + mock_ctrl.RotateToken.assert_awaited_once() + + +@pytest.mark.anyio +async def test_exporter_rotate_token_sync_callback(): + mock_channel = MagicMock(spec=grpc.aio.Channel) + mock_channel.close = AsyncMock() + + mock_ctrl = AsyncMock() + mock_ctrl.RotateToken.return_value = jumpstarter_pb2.RotateTokenResponse(token="rotated-token-value") + + async def channel_factory(): + return mock_channel + + recorded_tokens = [] + + def sync_on_rotated(token: str) -> None: + recorded_tokens.append(token) + + exporter = Exporter( + exporter_name="test-exporter", + token="old-token", + channel_factory=channel_factory, + device_factory=lambda: MagicMock(), + on_token_rotated=sync_on_rotated, + ) + + with patch("jumpstarter.exporter.exporter.jumpstarter_pb2_grpc.ControllerServiceStub", return_value=mock_ctrl): + token = await exporter.rotate_token() + + assert token == "rotated-token-value" + assert exporter.token == "rotated-token-value" + assert recorded_tokens == ["rotated-token-value"] + mock_ctrl.RotateToken.assert_awaited_once() + + +@pytest.mark.anyio +async def test_exporter_rotate_token_callback_failure(): + mock_channel = MagicMock(spec=grpc.aio.Channel) + mock_channel.close = AsyncMock() + + mock_ctrl = AsyncMock() + mock_ctrl.RotateToken.return_value = jumpstarter_pb2.RotateTokenResponse(token="rotated-token-value") + + async def channel_factory(): + return mock_channel + + on_rotated = AsyncMock(side_effect=RuntimeError("disk full or save failed")) + + exporter = Exporter( + exporter_name="test-exporter", + token="old-token", + channel_factory=channel_factory, + device_factory=lambda: MagicMock(), + on_token_rotated=on_rotated, + ) + + with patch("jumpstarter.exporter.exporter.jumpstarter_pb2_grpc.ControllerServiceStub", return_value=mock_ctrl): + token = await exporter.rotate_token() + + # Normal rotation result is preserved and token is updated + assert token == "rotated-token-value" + assert exporter.token == "rotated-token-value" + on_rotated.assert_awaited_once_with("rotated-token-value") + mock_ctrl.RotateToken.assert_awaited_once() + + +@pytest.mark.anyio +async def test_exporter_token_refresh_loop_near_expiry(): + # Token expiring in 1 second, with lead time 10s -> refresh sleep is 0s (immediate) + now = time.time() + token = _make_jwt({"iat": now - 100, "exp": now + 1}) + + mock_channel = MagicMock(spec=grpc.aio.Channel) + mock_channel.close = AsyncMock() + + mock_ctrl = AsyncMock() + # Next token valid for 1000s so it doesn't loop infinitely + next_token = _make_jwt({"iat": now, "exp": now + 100000}) + mock_ctrl.RotateToken.return_value = jumpstarter_pb2.RotateTokenResponse(token=next_token) + + async def channel_factory(): + return mock_channel + + exporter = Exporter( + exporter_name="test-exporter", + token=token, + channel_factory=channel_factory, + device_factory=lambda: MagicMock(), + token_refresh_lead_time=10.0, + ) + + with patch("jumpstarter.exporter.exporter.jumpstarter_pb2_grpc.ControllerServiceStub", return_value=mock_ctrl): + with anyio.move_on_after(0.5) as cancel_scope: + await exporter._token_refresh_loop() + + assert cancel_scope.cancel_called + assert exporter.token == next_token + mock_ctrl.RotateToken.assert_awaited_once() + + +@pytest.mark.anyio +async def test_exporter_token_refresh_loop_unimplemented(): + now = time.time() + token = _make_jwt({"iat": now - 100, "exp": now + 1}) + + mock_channel = MagicMock(spec=grpc.aio.Channel) + mock_channel.close = AsyncMock() + + mock_ctrl = AsyncMock() + mock_ctrl.RotateToken.side_effect = FakeRpcError(grpc.StatusCode.UNIMPLEMENTED) + + async def channel_factory(): + return mock_channel + + exporter = Exporter( + exporter_name="test-exporter", + token=token, + channel_factory=channel_factory, + device_factory=lambda: MagicMock(), + token_refresh_lead_time=10.0, + ) + + with patch("jumpstarter.exporter.exporter.jumpstarter_pb2_grpc.ControllerServiceStub", return_value=mock_ctrl): + # Should gracefully return without hanging or raising + with anyio.fail_after(2.0): + await exporter._token_refresh_loop() + + # Token was not updated + assert exporter.token == token + mock_ctrl.RotateToken.assert_awaited_once() + + +@pytest.mark.anyio +async def test_exporter_token_refresh_loop_standalone_and_no_token(): + exporter_standalone = Exporter( + exporter_name="test-exporter", + token="dummy-token", + channel_factory=AsyncMock(), + device_factory=lambda: MagicMock(), + ) + exporter_standalone._standalone = True + # Should return immediately + with anyio.fail_after(1.0): + await exporter_standalone._token_refresh_loop() + + exporter_no_token = Exporter( + exporter_name="test-exporter", + token="", + channel_factory=AsyncMock(), + device_factory=lambda: MagicMock(), + ) + # Should return immediately + with anyio.fail_after(1.0): + await exporter_no_token._token_refresh_loop() + + +@pytest.mark.anyio +async def test_config_on_token_rotated_saves_to_path(tmp_path): + config_file = tmp_path / "exporter.yaml" + cfg = ExporterConfigV1Alpha1( + metadata=ObjectMeta(name="my-exporter", namespace="default"), + endpoint="localhost:8082", + token="initial-token", + tls=TLSConfigV1Alpha1(insecure=True), + ) + ExporterConfigV1Alpha1.save(cfg, path=str(config_file)) + loaded = ExporterConfigV1Alpha1.load_path(config_file) + assert loaded.token == "initial-token" + + # Simulate create_exporter wiring + async with loaded.create_exporter() as exporter: + assert exporter.on_token_rotated is not None + await exporter.on_token_rotated("new-saved-token") + + assert loaded.token == "new-saved-token" + # Verify persisted to disk + reloaded = ExporterConfigV1Alpha1.load_path(config_file) + assert reloaded.token == "new-saved-token" diff --git a/python/packages/jumpstarter/jumpstarter/exporter/token_refresh.py b/python/packages/jumpstarter/jumpstarter/exporter/token_refresh.py new file mode 100644 index 000000000..c8542ff17 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/exporter/token_refresh.py @@ -0,0 +1,132 @@ +import base64 +import json +import logging +import time +from typing import Any + +logger = logging.getLogger(__name__) + +# Default maximum lead time before expiry to attempt token refresh (7 days). +# Matches the warning threshold used across Jumpstarter. +DEFAULT_TOKEN_REFRESH_LEAD_TIME: float = 7 * 24 * 3600.0 + +# Default fraction of total lifetime at which to trigger token refresh (20%). +DEFAULT_TOKEN_REFRESH_FRACTION: float = 0.2 + +# Minimum lead time before expiry to attempt token refresh (60 seconds). +MIN_TOKEN_REFRESH_LEAD_TIME: float = 60.0 + + +def decode_jwt_payload(token: str) -> dict[str, Any]: + """Decode the JSON payload from an unverified JWT token. + + Args: + token: JWT string (header.payload.signature) + + Returns: + Decoded payload dictionary + + Raises: + ValueError: If the token format is invalid or cannot be decoded + """ + parts = token.split(".") + if len(parts) != 3: + raise ValueError(f"Invalid JWT format: expected 3 parts, got {len(parts)}") + + payload_b64 = parts[1] + # Handle base64 padding + padded = payload_b64 + "=" * (-len(payload_b64) % 4) + try: + payload_bytes = base64.urlsafe_b64decode(padded) + return json.loads(payload_bytes) + except Exception as e: + raise ValueError(f"Failed to decode JWT payload: {e}") from e + + +def get_token_expiry(token: str) -> float | None: + """Extract expiry timestamp (Unix epoch seconds) from a JWT token. + + Returns: + float timestamp if 'exp' claim is present and valid, otherwise None. + """ + try: + payload = decode_jwt_payload(token) + except ValueError: + return None + + exp = payload.get("exp") + if isinstance(exp, (int, float)): + return float(exp) + return None + + +def get_token_remaining_seconds(token: str, now: float | None = None) -> float | None: + """Calculate the seconds remaining until token expiration. + + Args: + token: JWT string + now: Optional current timestamp (defaults to time.time()) + + Returns: + Remaining seconds (positive if valid, negative if expired), or None if no exp claim. + """ + exp = get_token_expiry(token) + if exp is None: + return None + if now is None: + now = time.time() + return exp - now + + +def calculate_token_refresh_sleep( + token: str, + now: float | None = None, + lead_time: float = DEFAULT_TOKEN_REFRESH_LEAD_TIME, + fraction: float = DEFAULT_TOKEN_REFRESH_FRACTION, + min_lead_time: float = MIN_TOKEN_REFRESH_LEAD_TIME, +) -> float | None: + """Calculate the number of seconds to sleep before requesting a new token. + + When the token has expired or is within the refresh threshold, returns 0.0 + indicating a refresh should be attempted immediately. + + Args: + token: JWT string + now: Optional current timestamp (defaults to time.time()) + lead_time: Maximum lead time before expiry to refresh (default: 7 days) + fraction: Fraction of total lifetime remaining to trigger refresh (default: 0.2) + min_lead_time: Minimum lead time before expiry to attempt refresh (default: 60s) + + Returns: + float >= 0: Seconds to sleep before refreshing (0.0 means refresh now) + None: Token is invalid or does not have an 'exp' claim (no refresh needed) + """ + if now is None: + now = time.time() + + try: + payload = decode_jwt_payload(token) + except ValueError: + return None + + exp = payload.get("exp") + if exp is None or not isinstance(exp, (int, float)): + return None + exp_f = float(exp) + + remaining = exp_f - now + if remaining <= 0: + return 0.0 + + iat = payload.get("iat") + if iat is not None and isinstance(iat, (int, float)) and float(iat) < exp_f: + total_lifetime = exp_f - float(iat) + lead = min(lead_time, max(total_lifetime * fraction, min_lead_time)) + # Ensure lead time never exceeds half the token's total lifetime + lead = min(lead, total_lifetime * 0.5) + else: + lead = min(lead_time, max(remaining * fraction, min_lead_time)) + lead = min(lead, remaining * 0.5) + + sleep_seconds = remaining - lead + return max(0.0, sleep_seconds) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/token_refresh_test.py b/python/packages/jumpstarter/jumpstarter/exporter/token_refresh_test.py new file mode 100644 index 000000000..5f8dd5515 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/exporter/token_refresh_test.py @@ -0,0 +1,120 @@ +import base64 +import json + +import pytest + +from jumpstarter.exporter.token_refresh import ( + calculate_token_refresh_sleep, + decode_jwt_payload, + get_token_expiry, + get_token_remaining_seconds, +) + + +def _make_jwt(payload: dict) -> str: + header = {"alg": "ES256", "typ": "JWT"} + h_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).rstrip(b"=").decode() + p_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=").decode() + return f"{h_b64}.{p_b64}.dummy_signature" + + +def test_decode_jwt_payload_valid(): + payload = {"sub": "exporter-1", "exp": 1234567890} + token = _make_jwt(payload) + decoded = decode_jwt_payload(token) + assert decoded["sub"] == "exporter-1" + assert decoded["exp"] == 1234567890 + + +def test_decode_jwt_payload_invalid(): + with pytest.raises(ValueError, match="expected 3 parts"): + decode_jwt_payload("invalid.token") + + with pytest.raises(ValueError, match="Failed to decode"): + decode_jwt_payload("part1.!!!notbase64!!!.part3") + + +def test_get_token_expiry(): + token = _make_jwt({"sub": "test", "exp": 1700000000}) + assert get_token_expiry(token) == 1700000000.0 + + token_no_exp = _make_jwt({"sub": "test"}) + assert get_token_expiry(token_no_exp) is None + + assert get_token_expiry("invalid") is None + + +def test_get_token_remaining_seconds(): + now = 1000.0 + token = _make_jwt({"exp": 1500}) + assert get_token_remaining_seconds(token, now=now) == 500.0 + + expired_token = _make_jwt({"exp": 900}) + assert get_token_remaining_seconds(expired_token, now=now) == -100.0 + + +def test_calculate_token_refresh_sleep_365_days(): + # 365 days token: 31,536,000s + now = 1_000_000.0 + iat = now + exp = now + 365 * 24 * 3600 + token = _make_jwt({"iat": iat, "exp": exp}) + + # Lead time is capped by DEFAULT_TOKEN_REFRESH_LEAD_TIME (7 days = 604,800s) + # Expected sleep: 365 days - 7 days = 358 days + sleep_s = calculate_token_refresh_sleep(token, now=now) + assert sleep_s is not None + assert pytest.approx(sleep_s, rel=1e-3) == (365 - 7) * 24 * 3600 + + +def test_calculate_token_refresh_sleep_fraction(): + # 10 days token: 20% is 2 days (172,800s), which is < 7 days + now = 1_000_000.0 + iat = now + exp = now + 10 * 24 * 3600 + token = _make_jwt({"iat": iat, "exp": exp}) + + # Lead time = 2 days + # Expected sleep: 8 days + sleep_s = calculate_token_refresh_sleep(token, now=now) + assert sleep_s is not None + assert pytest.approx(sleep_s, rel=1e-3) == 8 * 24 * 3600 + + +def test_calculate_token_refresh_sleep_short_token(): + # 100 seconds token: 50% cap applies (lead = 50s) + now = 1_000_000.0 + iat = now + exp = now + 100 + token = _make_jwt({"iat": iat, "exp": exp}) + + # Expected sleep: 50s + sleep_s = calculate_token_refresh_sleep(token, now=now) + assert sleep_s is not None + assert pytest.approx(sleep_s, rel=1e-3) == 50.0 + + +def test_calculate_token_refresh_sleep_near_expiry(): + # Token expiring in 3 days (less than 7 days lead time on a 365-day token) + now = 1_000_000.0 + iat = now - 362 * 24 * 3600 + exp = now + 3 * 24 * 3600 + token = _make_jwt({"iat": iat, "exp": exp}) + + sleep_s = calculate_token_refresh_sleep(token, now=now) + assert sleep_s == 0.0 + + +def test_calculate_token_refresh_sleep_expired(): + now = 1_000_000.0 + exp = now - 100 + token = _make_jwt({"exp": exp}) + + sleep_s = calculate_token_refresh_sleep(token, now=now) + assert sleep_s == 0.0 + + +def test_calculate_token_refresh_sleep_no_exp_or_invalid(): + token_no_exp = _make_jwt({"sub": "test"}) + assert calculate_token_refresh_sleep(token_no_exp) is None + assert calculate_token_refresh_sleep("not-a-token") is None From d365841a9ea8855b9158413c05b1c2ed4f5ea2e0 Mon Sep 17 00:00:00 2001 From: Benny Zlotnik Date: Thu, 3 Sep 2026 10:59:48 +0300 Subject: [PATCH 2/2] feat(controller): track exporter token expiry in status Add TokenExpiresAt timestamp and TokenExpiring condition to Exporter status, matching the Client status implementation. This enables downstream systems (Prometheus, alerting, kubectl) to detect exporter tokens approaching expiry before connectivity is lost. Also emit CredentialCreated and TokenExpiringSoon events, display Token Expires in kubectl get exporters, and ensure ExporterReconciler periodically requeues to track expiry even when an exporter is offline. Signed-off-by: Benny Zlotnik --- controller/api/v1alpha1/exporter_types.go | 8 +- .../api/v1alpha1/zz_generated.deepcopy.go | 4 + .../crd/bases/jumpstarter.dev_exporters.yaml | 8 ++ .../controller/exporter_controller.go | 36 +++++- .../controller/exporter_controller_test.go | 121 +++++++++++++++++- 5 files changed, 167 insertions(+), 10 deletions(-) diff --git a/controller/api/v1alpha1/exporter_types.go b/controller/api/v1alpha1/exporter_types.go index b2b87eb32..77222091d 100644 --- a/controller/api/v1alpha1/exporter_types.go +++ b/controller/api/v1alpha1/exporter_types.go @@ -52,13 +52,16 @@ type ExporterStatus struct { ExporterStatusValue string `json:"exporterStatus,omitempty"` // StatusMessage is an optional human-readable message describing the current state StatusMessage string `json:"statusMessage,omitempty"` + // TokenExpiresAt is the expiry time of the internal credential token. + TokenExpiresAt *metav1.Time `json:"tokenExpiresAt,omitempty"` } type ExporterConditionType string const ( - ExporterConditionTypeRegistered ExporterConditionType = "Registered" - ExporterConditionTypeOnline ExporterConditionType = "Online" + ExporterConditionTypeRegistered ExporterConditionType = "Registered" + ExporterConditionTypeOnline ExporterConditionType = "Online" + ExporterConditionTypeTokenExpiring ExporterConditionType = "TokenExpiring" ) // ExporterStatus values - PascalCase for Kubernetes, converted from proto ALL_CAPS @@ -78,6 +81,7 @@ const ( // +kubebuilder:printcolumn:name="Enabled",type="boolean",JSONPath=".spec.enabled" // +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.exporterStatus" // +kubebuilder:printcolumn:name="Message",type="string",JSONPath=".status.statusMessage",priority=1 +// +kubebuilder:printcolumn:name="Token Expires",type="string",JSONPath=".status.tokenExpiresAt" // Exporter is the Schema for the exporters API type Exporter struct { diff --git a/controller/api/v1alpha1/zz_generated.deepcopy.go b/controller/api/v1alpha1/zz_generated.deepcopy.go index bed80e9f3..113ee91c7 100644 --- a/controller/api/v1alpha1/zz_generated.deepcopy.go +++ b/controller/api/v1alpha1/zz_generated.deepcopy.go @@ -405,6 +405,10 @@ func (in *ExporterStatus) DeepCopyInto(out *ExporterStatus) { **out = **in } in.LastSeen.DeepCopyInto(&out.LastSeen) + if in.TokenExpiresAt != nil { + in, out := &in.TokenExpiresAt, &out.TokenExpiresAt + *out = (*in).DeepCopy() + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ExporterStatus. diff --git a/controller/deploy/operator/config/crd/bases/jumpstarter.dev_exporters.yaml b/controller/deploy/operator/config/crd/bases/jumpstarter.dev_exporters.yaml index 61fe9494c..cb55f08a8 100644 --- a/controller/deploy/operator/config/crd/bases/jumpstarter.dev_exporters.yaml +++ b/controller/deploy/operator/config/crd/bases/jumpstarter.dev_exporters.yaml @@ -25,6 +25,9 @@ spec: name: Message priority: 1 type: string + - jsonPath: .status.tokenExpiresAt + name: Token Expires + type: string name: v1alpha1 schema: openAPIV3Schema: @@ -203,6 +206,11 @@ spec: description: StatusMessage is an optional human-readable message describing the current state type: string + tokenExpiresAt: + description: TokenExpiresAt is the expiry time of the internal credential + token. + format: date-time + type: string type: object type: object served: true diff --git a/controller/internal/controller/exporter_controller.go b/controller/internal/controller/exporter_controller.go index 50726db4d..131930a0c 100644 --- a/controller/internal/controller/exporter_controller.go +++ b/controller/internal/controller/exporter_controller.go @@ -74,8 +74,11 @@ func (r *ExporterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c prevOnline := meta.IsStatusConditionTrue(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeOnline)) prevRegistered := meta.IsStatusConditionTrue(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeRegistered)) + prevCredential := exporter.Status.Credential + prevTokenExpiring := meta.IsStatusConditionTrue(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeTokenExpiring)) - if err := r.reconcileStatusCredential(ctx, &exporter); err != nil { + tokenExpiry, err := r.reconcileStatusCredential(ctx, &exporter) + if err != nil { return ctrl.Result{}, err } @@ -92,6 +95,14 @@ func (r *ExporterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c return ctrl.Result{}, err } + reconcileTokenExpiry( + &exporter.Status.Conditions, + &exporter.Status.TokenExpiresAt, + tokenExpiry, + exporter.Generation, + string(jumpstarterdevv1alpha1.ExporterConditionTypeTokenExpiring), + ) + newOnline := meta.IsStatusConditionTrue(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeOnline)) newRegistered := meta.IsStatusConditionTrue(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeRegistered)) @@ -100,6 +111,17 @@ func (r *ExporterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c } // Emit only after status patch succeeds. + if prevCredential == nil && exporter.Status.Credential != nil { + r.emitEventf(&exporter, corev1.EventTypeNormal, "CredentialCreated", + "Credential secret created for exporter: secret=%s", exporter.Status.Credential.Name) + } + + newTokenExpiring := meta.IsStatusConditionTrue(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeTokenExpiring)) + if !prevTokenExpiring && newTokenExpiring { + r.emitEventf(&exporter, corev1.EventTypeWarning, "TokenExpiringSoon", + "Token for exporter %s is expiring soon: expires=%s", exporter.Name, tokenExpiry.UTC().Format(time.RFC3339)) + } + if !prevRegistered && newRegistered { r.emitEventf(&exporter, corev1.EventTypeNormal, "ExporterRegistered", "Exporter registered its capabilities: deviceCount=%d", len(exporter.Status.Devices)) @@ -125,24 +147,28 @@ func (r *ExporterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c } } + if result.RequeueAfter == 0 || result.RequeueAfter > tokenExpiryRequeueInterval { + result.RequeueAfter = tokenExpiryRequeueInterval + } + return result, nil } func (r *ExporterReconciler) reconcileStatusCredential( ctx context.Context, exporter *jumpstarterdevv1alpha1.Exporter, -) error { - secret, _, err := ensureSecret(ctx, client.ObjectKey{ +) (time.Time, error) { + secret, expiry, err := ensureSecret(ctx, client.ObjectKey{ Name: exporter.Name + "-exporter", Namespace: exporter.Namespace, }, r.Client, r.Scheme, r.Signer, exporter.InternalSubject(), exporter) if err != nil { - return fmt.Errorf("reconcileStatusCredential: failed to prepare credential for exporter: %w", err) + return time.Time{}, fmt.Errorf("reconcileStatusCredential: failed to prepare credential for exporter: %w", err) } exporter.Status.Credential = &corev1.LocalObjectReference{ Name: secret.Name, } - return nil + return expiry, nil } func (r *ExporterReconciler) reconcileStatusLeaseRef( diff --git a/controller/internal/controller/exporter_controller_test.go b/controller/internal/controller/exporter_controller_test.go index d42b6ea21..ab43d8fb0 100644 --- a/controller/internal/controller/exporter_controller_test.go +++ b/controller/internal/controller/exporter_controller_test.go @@ -18,11 +18,13 @@ package controller import ( "context" + "time" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -88,12 +90,20 @@ var _ = Describe("Exporter Controller", func() { Signer: signer, } - _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{ + res, err := controllerReconciler.Reconcile(ctx, reconcile.Request{ NamespacedName: typeNamespacedName, }) Expect(err).NotTo(HaveOccurred()) - // TODO(user): Add more specific assertions depending on your controller's reconciliation logic. - // Example: If you expect a certain status condition after reconciliation, verify it here. + Expect(res.RequeueAfter).To(Equal(tokenExpiryRequeueInterval)) + + exporter := &jumpstarterdevv1alpha1.Exporter{} + Expect(k8sClient.Get(ctx, typeNamespacedName, exporter)).To(Succeed()) + Expect(exporter.Status.TokenExpiresAt).NotTo(BeNil()) + + cond := meta.FindStatusCondition(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeTokenExpiring)) + Expect(cond).NotTo(BeNil()) + Expect(cond.Status).To(Equal(metav1.ConditionFalse)) + Expect(cond.Reason).To(Equal("Valid")) }) It("should reconcile a missing token secret", func() { By("recreating the secret") @@ -125,5 +135,110 @@ var _ = Describe("Exporter Controller", func() { Name: resourceName + "-exporter", }, secret)).To(Succeed()) }) + It("should report TokenExpiring condition when token is expiring soon", func() { + By("reconciling an exporter with a near-expiry token") + signer, err := oidc.NewSignerFromSeed([]byte{}, "https://example.com", "dummy") + Expect(err).NotTo(HaveOccurred()) + + controllerReconciler := &ExporterReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Signer: signer, + } + + // First reconcile to create the secret + _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: typeNamespacedName, + }) + Expect(err).NotTo(HaveOccurred()) + + // Modify secret with near-expiry token (5 days remaining) + nearExpirySigner, err := oidc.NewSignerFromSeed([]byte{}, "https://example.com", "dummy") + Expect(err).NotTo(HaveOccurred()) + nearExpirySigner.SetTokenLifetime(5 * 24 * time.Hour) + nearExpiryToken, err := nearExpirySigner.Token("exporter-test") + Expect(err).NotTo(HaveOccurred()) + + secret := &corev1.Secret{} + Expect(k8sClient.Get(ctx, types.NamespacedName{ + Namespace: "default", + Name: resourceName + "-exporter", + }, secret)).To(Succeed()) + + secret.Data[TokenKey] = []byte(nearExpiryToken) + Expect(k8sClient.Update(ctx, secret)).To(Succeed()) + + // Reconcile again + _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: typeNamespacedName, + }) + Expect(err).NotTo(HaveOccurred()) + + // Verify secret token remains intact so alerts can fire + Expect(k8sClient.Get(ctx, types.NamespacedName{ + Namespace: "default", + Name: resourceName + "-exporter", + }, secret)).To(Succeed()) + Expect(string(secret.Data[TokenKey])).To(Equal(nearExpiryToken)) + + // Verify exporter status condition is True with ExpiresWithin7Days + exporter := &jumpstarterdevv1alpha1.Exporter{} + Expect(k8sClient.Get(ctx, typeNamespacedName, exporter)).To(Succeed()) + Expect(exporter.Status.TokenExpiresAt).NotTo(BeNil()) + + cond := meta.FindStatusCondition(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeTokenExpiring)) + Expect(cond).NotTo(BeNil()) + Expect(cond.Status).To(Equal(metav1.ConditionTrue)) + Expect(cond.Reason).To(Equal("ExpiresWithin7Days")) + }) + + It("should reconcile an invalid token secret", func() { + By("recreating an invalid secret") + signer, err := oidc.NewSignerFromSeed([]byte{}, "https://example.com", "dummy") + Expect(err).NotTo(HaveOccurred()) + + controllerReconciler := &ExporterReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Signer: signer, + } + + // First reconcile to create the secret + _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: typeNamespacedName, + }) + Expect(err).NotTo(HaveOccurred()) + + // Corrupt the secret + secret := &corev1.Secret{} + Expect(k8sClient.Get(ctx, types.NamespacedName{ + Namespace: "default", + Name: resourceName + "-exporter", + }, secret)).To(Succeed()) + + secret.Data[TokenKey] = []byte("invalid-token") + Expect(k8sClient.Update(ctx, secret)).To(Succeed()) + + // Reconcile + _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: typeNamespacedName, + }) + Expect(err).NotTo(HaveOccurred()) + + // Verify secret was recreated with a valid token + Expect(k8sClient.Get(ctx, types.NamespacedName{ + Namespace: "default", + Name: resourceName + "-exporter", + }, secret)).To(Succeed()) + Expect(string(secret.Data[TokenKey])).NotTo(Equal("invalid-token")) + + // Verify condition is Valid + exporter := &jumpstarterdevv1alpha1.Exporter{} + Expect(k8sClient.Get(ctx, typeNamespacedName, exporter)).To(Succeed()) + cond := meta.FindStatusCondition(exporter.Status.Conditions, string(jumpstarterdevv1alpha1.ExporterConditionTypeTokenExpiring)) + Expect(cond).NotTo(BeNil()) + Expect(cond.Status).To(Equal(metav1.ConditionFalse)) + Expect(cond.Reason).To(Equal("Valid")) + }) }) })