404 lines
12 KiB
Go
404 lines
12 KiB
Go
package localcontrol
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/binary"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"strings"
|
|
"unicode/utf8"
|
|
|
|
iop "iop/proto/gen/iop"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
"google.golang.org/protobuf/reflect/protoreflect"
|
|
)
|
|
|
|
const (
|
|
ProtocolVersion uint32 = 1
|
|
MaxFrameBytes = 1 << 20
|
|
|
|
maxIdentifierBytes = 256
|
|
maxOperationBytes = 64
|
|
maxSafeTextBytes = 1024
|
|
)
|
|
|
|
const (
|
|
OperationRuntimeStatus = "runtime.status"
|
|
OperationProjectStatus = "project.status"
|
|
OperationOverlayStatus = "overlay.status"
|
|
OperationIntegrationStatus = "integration.status"
|
|
OperationBlockerList = "blocker.list"
|
|
OperationProcessStatus = "process.status"
|
|
|
|
OperationProjectStart = "project.start"
|
|
OperationProjectStop = "project.stop"
|
|
OperationProjectResume = "project.resume"
|
|
|
|
OperationClientStart = "client.start"
|
|
OperationClientStop = "client.stop"
|
|
OperationClientFocus = "client.focus"
|
|
OperationClientDetail = "client.detail"
|
|
)
|
|
|
|
const (
|
|
ErrorMalformedFrame = "malformed_frame"
|
|
ErrorUnsupportedVersion = "unsupported_version"
|
|
ErrorUnsupportedOperation = "unsupported_operation"
|
|
ErrorInvalidState = "invalid_state"
|
|
ErrorPermissionDenied = "permission_denied"
|
|
ErrorCommandIDConflict = "command_id_conflict"
|
|
ErrorReplayUnavailable = "replay_unavailable"
|
|
ErrorInternal = "internal"
|
|
)
|
|
|
|
// ProtocolError is a bounded local-control failure. Detail from an underlying
|
|
// error is intentionally not retained because it can contain private paths or
|
|
// subprocess output.
|
|
type ProtocolError struct {
|
|
Code string
|
|
SafeMessage string
|
|
Retryable bool
|
|
ReplayFloor uint64
|
|
SnapshotRequired bool
|
|
SnapshotMarker string
|
|
}
|
|
|
|
func (e *ProtocolError) Error() string {
|
|
if e == nil {
|
|
return ""
|
|
}
|
|
return e.Code + ": " + e.SafeMessage
|
|
}
|
|
|
|
func protocolError(code, message string) *ProtocolError {
|
|
return &ProtocolError{
|
|
Code: code,
|
|
SafeMessage: boundedSafeText(message),
|
|
}
|
|
}
|
|
|
|
func malformed(message string) *ProtocolError {
|
|
return protocolError(ErrorMalformedFrame, message)
|
|
}
|
|
|
|
// ParseEnvelope decodes one bounded protobuf payload without discarding future
|
|
// fields. ValidateRequest rejects all retained unknown fields fail-closed.
|
|
func ParseEnvelope(payload []byte) (*iop.AgentLocalEnvelope, error) {
|
|
if len(payload) == 0 || len(payload) > MaxFrameBytes {
|
|
return nil, malformed("local-control payload size is invalid")
|
|
}
|
|
envelope := &iop.AgentLocalEnvelope{}
|
|
if err := (proto.UnmarshalOptions{DiscardUnknown: false}).Unmarshal(payload, envelope); err != nil {
|
|
return nil, malformed("local-control payload is not valid protobuf")
|
|
}
|
|
return envelope, nil
|
|
}
|
|
|
|
// ValidateRequest verifies version, kind, identifiers, operation, payload
|
|
// pairing, and unknown fields before any service port can be called.
|
|
func ValidateRequest(envelope *iop.AgentLocalEnvelope) *ProtocolError {
|
|
if envelope == nil {
|
|
return malformed("local-control envelope is required")
|
|
}
|
|
if proto.Size(envelope) <= 0 || proto.Size(envelope) > MaxFrameBytes {
|
|
return malformed("local-control envelope size is invalid")
|
|
}
|
|
if hasUnknownFields(envelope.ProtoReflect()) {
|
|
return malformed("unknown local-control fields are not accepted")
|
|
}
|
|
if envelope.GetProtocolVersion() != ProtocolVersion {
|
|
return protocolError(
|
|
ErrorUnsupportedVersion,
|
|
"local-control protocol version is not supported",
|
|
)
|
|
}
|
|
if envelope.GetKind() != iop.AgentLocalKind_AGENT_LOCAL_KIND_REQUEST {
|
|
return malformed("local-control envelope kind must be request")
|
|
}
|
|
if envelope.GetRequest() == nil {
|
|
return malformed("request payload does not match envelope kind")
|
|
}
|
|
if envelope.GetCorrelationId() != "" || envelope.GetEventSequence() != 0 {
|
|
return malformed("request envelope contains response-only fields")
|
|
}
|
|
if !validIdentifier(envelope.GetMessageId(), maxIdentifierBytes) {
|
|
return malformed("message_id is required and invalid")
|
|
}
|
|
if !validIdentifier(envelope.GetOperation(), maxOperationBytes) {
|
|
return malformed("operation is required and invalid")
|
|
}
|
|
|
|
request := envelope.GetRequest()
|
|
if request.ReplayAfterSequence != nil {
|
|
if !validIdentifier(request.GetReplayDaemonId(), maxIdentifierBytes) {
|
|
return malformed("replay_daemon_id is required with a replay cursor")
|
|
}
|
|
} else if request.GetReplayDaemonId() != "" {
|
|
return malformed("replay_daemon_id requires a replay cursor")
|
|
}
|
|
|
|
switch {
|
|
case isReadOperation(envelope.GetOperation()):
|
|
if request.GetRead() == nil || request.GetProject() != nil || request.GetClient() != nil {
|
|
return malformed("read operation requires a read payload")
|
|
}
|
|
if request.GetCommandId() != "" {
|
|
return malformed("read operation must not contain command_id")
|
|
}
|
|
return validateReadRequest(envelope.GetOperation(), request.GetRead())
|
|
case isProjectMutation(envelope.GetOperation()):
|
|
if request.GetProject() == nil || request.GetRead() != nil || request.GetClient() != nil {
|
|
return malformed("project mutation requires a project payload")
|
|
}
|
|
if !validIdentifier(request.GetCommandId(), maxIdentifierBytes) {
|
|
return malformed("project mutation requires a valid command_id")
|
|
}
|
|
return validateProjectRequest(envelope.GetOperation(), request.GetProject())
|
|
case isClientMutation(envelope.GetOperation()):
|
|
if request.GetClient() == nil || request.GetRead() != nil || request.GetProject() != nil {
|
|
return malformed("client mutation requires a client payload")
|
|
}
|
|
if !validIdentifier(request.GetCommandId(), maxIdentifierBytes) {
|
|
return malformed("client mutation requires a valid command_id")
|
|
}
|
|
if !validOptionalIdentifier(request.GetClient().GetClientKind()) ||
|
|
!validOptionalIdentifier(request.GetClient().GetCapability()) {
|
|
return malformed("client operation arguments are invalid")
|
|
}
|
|
return nil
|
|
default:
|
|
return protocolError(
|
|
ErrorUnsupportedOperation,
|
|
"local-control operation is not supported",
|
|
)
|
|
}
|
|
}
|
|
|
|
func validateReadRequest(
|
|
operation string,
|
|
request *iop.AgentLocalReadRequest,
|
|
) *ProtocolError {
|
|
if !validOptionalIdentifier(request.GetProjectId()) ||
|
|
!validOptionalIdentifier(request.GetWorkUnitId()) ||
|
|
!validOptionalIdentifier(request.GetClientKind()) {
|
|
return malformed("read operation arguments are invalid")
|
|
}
|
|
switch operation {
|
|
case OperationRuntimeStatus:
|
|
return nil
|
|
case OperationProjectStatus, OperationBlockerList:
|
|
if request.GetProjectId() == "" {
|
|
return malformed("project_id is required for this read operation")
|
|
}
|
|
case OperationOverlayStatus, OperationIntegrationStatus:
|
|
if request.GetProjectId() == "" || request.GetWorkUnitId() == "" {
|
|
return malformed("project_id and work_unit_id are required for this read operation")
|
|
}
|
|
case OperationProcessStatus:
|
|
if request.GetClientKind() == "" {
|
|
return malformed("client_kind is required for process.status")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateProjectRequest(
|
|
operation string,
|
|
request *iop.AgentLocalProjectRequest,
|
|
) *ProtocolError {
|
|
if !validIdentifier(request.GetProjectId(), maxIdentifierBytes) ||
|
|
!validOptionalIdentifier(request.GetWorkspaceId()) ||
|
|
!validOptionalIdentifier(request.GetMilestoneId()) {
|
|
return malformed("project operation arguments are invalid")
|
|
}
|
|
if operation == OperationProjectStart &&
|
|
(request.GetWorkspaceId() == "" || request.GetMilestoneId() == "") {
|
|
return malformed("project.start requires workspace_id and milestone_id")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func isReadOperation(operation string) bool {
|
|
switch operation {
|
|
case OperationRuntimeStatus,
|
|
OperationProjectStatus,
|
|
OperationOverlayStatus,
|
|
OperationIntegrationStatus,
|
|
OperationBlockerList,
|
|
OperationProcessStatus:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func isProjectMutation(operation string) bool {
|
|
switch operation {
|
|
case OperationProjectStart, OperationProjectStop, OperationProjectResume:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func isClientMutation(operation string) bool {
|
|
switch operation {
|
|
case OperationClientStart, OperationClientStop, OperationClientFocus, OperationClientDetail:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func validIdentifier(value string, limit int) bool {
|
|
return value != "" && validBoundedText(value, limit)
|
|
}
|
|
|
|
func validOptionalIdentifier(value string) bool {
|
|
return value == "" || validBoundedText(value, maxIdentifierBytes)
|
|
}
|
|
|
|
func validBoundedText(value string, limit int) bool {
|
|
if len(value) > limit || !utf8.ValidString(value) || strings.TrimSpace(value) != value {
|
|
return false
|
|
}
|
|
for _, char := range value {
|
|
if char < 0x20 || char == 0x7f {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func boundedSafeText(value string) string {
|
|
if !utf8.ValidString(value) {
|
|
return "local-control request failed"
|
|
}
|
|
value = strings.Map(func(char rune) rune {
|
|
if char < 0x20 || char == 0x7f {
|
|
return -1
|
|
}
|
|
return char
|
|
}, value)
|
|
if len(value) > maxSafeTextBytes {
|
|
value = value[:maxSafeTextBytes]
|
|
for !utf8.ValidString(value) {
|
|
value = value[:len(value)-1]
|
|
}
|
|
}
|
|
if value == "" {
|
|
return "local-control request failed"
|
|
}
|
|
return value
|
|
}
|
|
|
|
func hasUnknownFields(message protoreflect.Message) bool {
|
|
if len(message.GetUnknown()) != 0 {
|
|
return true
|
|
}
|
|
found := false
|
|
message.Range(func(field protoreflect.FieldDescriptor, value protoreflect.Value) bool {
|
|
if field.IsMap() {
|
|
if field.MapValue().Kind() != protoreflect.MessageKind {
|
|
return true
|
|
}
|
|
value.Map().Range(func(_ protoreflect.MapKey, entry protoreflect.Value) bool {
|
|
if hasUnknownFields(entry.Message()) {
|
|
found = true
|
|
return false
|
|
}
|
|
return true
|
|
})
|
|
return !found
|
|
}
|
|
if field.IsList() {
|
|
if field.Kind() != protoreflect.MessageKind {
|
|
return true
|
|
}
|
|
list := value.List()
|
|
for index := 0; index < list.Len(); index++ {
|
|
if hasUnknownFields(list.Get(index).Message()) {
|
|
found = true
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
if field.Kind() == protoreflect.MessageKind &&
|
|
hasUnknownFields(value.Message()) {
|
|
found = true
|
|
return false
|
|
}
|
|
return true
|
|
})
|
|
return found
|
|
}
|
|
|
|
func immutableCommandHash(
|
|
operation string,
|
|
request *iop.AgentLocalRequest,
|
|
) (string, error) {
|
|
if request == nil {
|
|
return "", fmt.Errorf("request is required")
|
|
}
|
|
immutable := proto.Clone(request).(*iop.AgentLocalRequest)
|
|
immutable.CommandId = ""
|
|
immutable.ReplayDaemonId = ""
|
|
immutable.ReplayAfterSequence = nil
|
|
payload, err := (proto.MarshalOptions{Deterministic: true}).Marshal(immutable)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
hash := sha256.New()
|
|
var length [8]byte
|
|
binary.BigEndian.PutUint64(length[:], uint64(len(operation)))
|
|
_, _ = hash.Write(length[:])
|
|
_, _ = hash.Write([]byte(operation))
|
|
binary.BigEndian.PutUint64(length[:], uint64(len(payload)))
|
|
_, _ = hash.Write(length[:])
|
|
_, _ = hash.Write(payload)
|
|
return "sha256:" + hex.EncodeToString(hash.Sum(nil)), nil
|
|
}
|
|
|
|
func derivedMessageID(prefix, source string) string {
|
|
sum := sha256.Sum256([]byte(source))
|
|
return prefix + ":" + hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
func errorEnvelope(
|
|
request *iop.AgentLocalEnvelope,
|
|
failure *ProtocolError,
|
|
) *iop.AgentLocalEnvelope {
|
|
if failure == nil {
|
|
failure = protocolError(ErrorInternal, "local-control request failed")
|
|
}
|
|
correlationID := ""
|
|
operation := ""
|
|
messageID := "error:uncorrelated"
|
|
if request != nil {
|
|
correlationID = request.GetMessageId()
|
|
operation = request.GetOperation()
|
|
if correlationID != "" {
|
|
messageID = derivedMessageID("error", correlationID)
|
|
}
|
|
}
|
|
return &iop.AgentLocalEnvelope{
|
|
ProtocolVersion: ProtocolVersion,
|
|
Kind: iop.AgentLocalKind_AGENT_LOCAL_KIND_ERROR,
|
|
MessageId: messageID,
|
|
CorrelationId: correlationID,
|
|
Operation: operation,
|
|
Payload: &iop.AgentLocalEnvelope_Error{
|
|
Error: &iop.AgentLocalError{
|
|
Code: failure.Code,
|
|
SafeMessage: boundedSafeText(failure.SafeMessage),
|
|
Retryable: failure.Retryable,
|
|
CorrelationId: correlationID,
|
|
ReplayFloor: failure.ReplayFloor,
|
|
SnapshotRequired: failure.SnapshotRequired,
|
|
SnapshotMarker: failure.SnapshotMarker,
|
|
},
|
|
},
|
|
}
|
|
}
|