381 lines
11 KiB
Go
381 lines
11 KiB
Go
package rabbitproxy
|
|
|
|
import (
|
|
"context"
|
|
"crypto/subtle"
|
|
"crypto/tls"
|
|
"encoding/binary"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"linkport/internal/cluster"
|
|
"linkport/internal/model"
|
|
"linkport/internal/protocol/amqpprobe"
|
|
"linkport/internal/protocol/transport"
|
|
"net"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
type Handler struct{}
|
|
|
|
func New() *Handler { return &Handler{} }
|
|
|
|
var protocolHeader = []byte{'A', 'M', 'Q', 'P', 0, 0, 9, 1}
|
|
|
|
type frame struct {
|
|
kind byte
|
|
channel uint16
|
|
body []byte
|
|
}
|
|
|
|
func read(c net.Conn) (frame, error) {
|
|
var h [7]byte
|
|
if _, err := io.ReadFull(c, h[:]); err != nil {
|
|
return frame{}, err
|
|
}
|
|
n := binary.BigEndian.Uint32(h[3:])
|
|
if n > 16<<20 {
|
|
return frame{}, errors.New("AMQP frame too large")
|
|
}
|
|
b := make([]byte, int(n)+1)
|
|
if _, err := io.ReadFull(c, b); err != nil {
|
|
return frame{}, err
|
|
}
|
|
if b[n] != 206 {
|
|
return frame{}, errors.New("invalid AMQP frame ending")
|
|
}
|
|
if h[0] != 1 && h[0] != 2 && h[0] != 3 && h[0] != 8 {
|
|
return frame{}, errors.New("invalid AMQP frame type")
|
|
}
|
|
return frame{h[0], binary.BigEndian.Uint16(h[1:3]), b[:n]}, nil
|
|
}
|
|
func write(c io.Writer, f frame) error {
|
|
b := []byte{f.kind, byte(f.channel >> 8), byte(f.channel)}
|
|
b = binary.BigEndian.AppendUint32(b, uint32(len(f.body)))
|
|
b = append(b, f.body...)
|
|
b = append(b, 206)
|
|
return transport.Write(c, b)
|
|
}
|
|
func method(f frame, class, method uint16) bool {
|
|
return f.kind == 1 && len(f.body) >= 4 && binary.BigEndian.Uint16(f.body) == class && binary.BigEndian.Uint16(f.body[2:]) == method
|
|
}
|
|
func long(b []byte) ([]byte, []byte, error) {
|
|
if len(b) < 4 {
|
|
return nil, nil, io.ErrUnexpectedEOF
|
|
}
|
|
n := int(binary.BigEndian.Uint32(b))
|
|
if n > len(b)-4 {
|
|
return nil, nil, io.ErrUnexpectedEOF
|
|
}
|
|
return b[4 : 4+n], b[4+n:], nil
|
|
}
|
|
func short(b []byte) (string, []byte, error) {
|
|
if len(b) < 1 || int(b[0]) > len(b)-1 {
|
|
return "", nil, io.ErrUnexpectedEOF
|
|
}
|
|
return string(b[1 : 1+int(b[0])]), b[1+int(b[0]):], nil
|
|
}
|
|
func addLong(b, data []byte) []byte {
|
|
b = binary.BigEndian.AppendUint32(b, uint32(len(data)))
|
|
return append(b, data...)
|
|
}
|
|
func startOK(user, pass string) frame {
|
|
b := []byte{0, 10, 0, 11, 0, 0, 0, 0, 5, 'P', 'L', 'A', 'I', 'N'}
|
|
b = addLong(b, []byte("\x00"+user+"\x00"+pass))
|
|
b = append(b, 5, 'e', 'n', '_', 'U', 'S')
|
|
return frame{1, 0, b}
|
|
}
|
|
func start(c net.Conn) (frame, error) {
|
|
if err := transport.Write(c, protocolHeader); err != nil {
|
|
return frame{}, err
|
|
}
|
|
f, err := read(c)
|
|
if err != nil {
|
|
return f, err
|
|
}
|
|
if !method(f, 10, 10) || len(f.body) < 6 {
|
|
return f, errors.New("AMQP connection.start expected")
|
|
}
|
|
_, rest, err := long(f.body[6:])
|
|
if err != nil {
|
|
return f, err
|
|
}
|
|
mechanisms, _, err := long(rest)
|
|
if err != nil {
|
|
return f, err
|
|
}
|
|
if !strings.Contains(" "+string(mechanisms)+" ", " PLAIN ") {
|
|
return f, errors.New("AMQP target requires unsupported authentication")
|
|
}
|
|
return f, nil
|
|
}
|
|
func credentials(f frame) (string, string, error) {
|
|
if !method(f, 10, 11) {
|
|
return "", "", errors.New("AMQP start-ok expected")
|
|
}
|
|
_, rest, err := long(f.body[4:])
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
mechanism, rest, err := short(rest)
|
|
if err != nil || mechanism != "PLAIN" {
|
|
return "", "", errors.New("AMQP PLAIN required")
|
|
}
|
|
response, rest, err := long(rest)
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
_, rest, err = short(rest)
|
|
if err != nil || len(rest) != 0 {
|
|
return "", "", errors.New("invalid AMQP locale")
|
|
}
|
|
parts := strings.Split(string(response), "\x00")
|
|
if len(parts) != 3 || parts[0] != "" {
|
|
return "", "", errors.New("invalid AMQP credentials")
|
|
}
|
|
return parts[1], parts[2], nil
|
|
}
|
|
func target(ctx context.Context, p model.Proxy, dial model.DialFunc) (net.Conn, frame, error) {
|
|
if p.ClusterEnabled() {
|
|
for _, n := range append([]model.ClusterNode{{Host: p.TargetHost, Port: p.TargetPort}}, p.Cluster.Nodes...) {
|
|
candidate := p
|
|
candidate.Cluster = model.ClusterConfig{}
|
|
candidate.TargetHost, candidate.TargetPort = n.Host, n.Port
|
|
attempt, cancel := context.WithTimeout(ctx, 3*time.Second)
|
|
c, f, err := target(attempt, candidate, dial)
|
|
cancel()
|
|
if err == nil {
|
|
return c, f, nil
|
|
}
|
|
}
|
|
return nil, frame{}, errors.New("no reachable AMQP cluster seed")
|
|
}
|
|
c, err := transport.Dial(ctx, p, dial)
|
|
if err != nil {
|
|
return nil, frame{}, err
|
|
}
|
|
c.SetDeadline(time.Now().Add(15 * time.Second))
|
|
stop := context.AfterFunc(ctx, func() { c.Close() })
|
|
defer stop()
|
|
f, err := start(c)
|
|
if err != nil {
|
|
c.Close()
|
|
return nil, f, err
|
|
}
|
|
return c, f, nil
|
|
}
|
|
func (h *Handler) Test(ctx context.Context, p model.Proxy, dial model.DialFunc) (model.TestResult, error) {
|
|
p, _, err := cluster.Resolve(ctx, p, dial)
|
|
if err != nil {
|
|
return model.TestResult{}, err
|
|
}
|
|
vhost := p.AuthSource
|
|
if vhost == "" {
|
|
vhost = "/"
|
|
}
|
|
if len(vhost) > 255 {
|
|
return model.TestResult{}, errors.New("AMQP virtual host too long")
|
|
}
|
|
result, err := amqpprobe.Probe(ctx, p, dial)
|
|
if err != nil {
|
|
return model.TestResult{}, err
|
|
}
|
|
version := result.Version
|
|
if version == "" {
|
|
version = "AMQP 0-9-1 (server version unavailable)"
|
|
}
|
|
return model.TestResult{Version: version, TLS: p.TargetTLS, Compression: "none", Role: vhost}, nil
|
|
}
|
|
func (h *Handler) Serve(ctx context.Context, raw net.Conn, p model.Proxy, dial model.DialFunc, tc *tls.Config, event model.EventFunc) error {
|
|
defer raw.Close()
|
|
stop := context.AfterFunc(ctx, func() { raw.Close() })
|
|
defer stop()
|
|
c, err := transport.Accept(ctx, raw, tc)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
header := make([]byte, 8)
|
|
if _, err = io.ReadFull(c, header); err != nil {
|
|
return err
|
|
}
|
|
if string(header) != string(protocolHeader) {
|
|
return errors.New("AMQP 0-9-1 required")
|
|
}
|
|
var up net.Conn
|
|
var greeting frame
|
|
if p.HasAccountTargets() {
|
|
// Local SASL greeting lets the client choose its route before we dial.
|
|
body := addLong([]byte{0, 10, 0, 10, 0, 9}, nil)
|
|
body = addLong(body, []byte("PLAIN"))
|
|
body = addLong(body, []byte("en_US"))
|
|
greeting = frame{kind: 1, body: body}
|
|
} else {
|
|
up, greeting, err = target(ctx, p, dial)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
initial := up
|
|
defer initial.Close()
|
|
stopUp := context.AfterFunc(ctx, func() { initial.Close() })
|
|
defer stopUp()
|
|
}
|
|
// Restrict the downstream SASL choice to the authentication method we terminate.
|
|
properties, rest, err := long(greeting.body[6:])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, locales, err := long(rest)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
greeting.body = addLong([]byte{0, 10, 0, 10, 0, 9}, properties)
|
|
greeting.body = addLong(greeting.body, []byte("PLAIN"))
|
|
greeting.body = append(greeting.body, locales...)
|
|
if err = write(c, greeting); err != nil {
|
|
return err
|
|
}
|
|
auth, err := read(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
user, pass, err := credentials(auth)
|
|
account, exists := transport.Account(p, user, raw.RemoteAddr())
|
|
if err != nil || !exists || subtle.ConstantTimeCompare([]byte(pass), []byte(account.Password)) != 1 {
|
|
transport.AuthenticationEvent(event, user, false)
|
|
return errors.New("AMQP gateway authentication failed")
|
|
}
|
|
p = account.Apply(p)
|
|
dial = p.DialForTarget(dial)
|
|
transport.AuthenticationEvent(event, user, true)
|
|
if p.ClusterEnabled() || up == nil {
|
|
resolved, _, e := cluster.Resolve(ctx, p, dial)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
p = resolved
|
|
if up != nil {
|
|
up.Close()
|
|
}
|
|
up, _, err = target(ctx, p, dial)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer up.Close()
|
|
selected := up
|
|
stopSelected := context.AfterFunc(ctx, func() { selected.Close() })
|
|
defer stopSelected()
|
|
}
|
|
if err = write(up, startOK(p.TargetUser, p.TargetPassword)); err != nil {
|
|
return err
|
|
}
|
|
tune, err := read(up)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !method(tune, 10, 30) {
|
|
return errors.New("AMQP target authentication failed")
|
|
}
|
|
if err = write(c, tune); err != nil {
|
|
return err
|
|
}
|
|
accepted, err := read(c)
|
|
if err != nil || !method(accepted, 10, 31) {
|
|
return errors.New("AMQP tune-ok expected")
|
|
}
|
|
if err = write(up, accepted); err != nil {
|
|
return err
|
|
}
|
|
opened, err := read(c)
|
|
if err != nil || !method(opened, 10, 40) {
|
|
return errors.New("AMQP open expected")
|
|
}
|
|
vhost, _, err := short(opened.body[4:])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = model.AuditCommand(ctx, "connection.open", vhost, "allowed"); err != nil {
|
|
return err
|
|
}
|
|
if err = write(up, opened); err != nil {
|
|
return err
|
|
}
|
|
response, err := read(up)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = write(c, response); err != nil {
|
|
return err
|
|
}
|
|
if !method(response, 10, 41) {
|
|
return errors.New("AMQP virtual host unavailable")
|
|
}
|
|
raw.SetDeadline(time.Time{})
|
|
up.SetDeadline(time.Time{})
|
|
out := &transport.LockedWriter{Writer: c}
|
|
var timing rpcTimings
|
|
go func() {
|
|
defer raw.Close()
|
|
for {
|
|
f, e := read(up)
|
|
if e != nil {
|
|
return
|
|
}
|
|
if write(out, f) != nil {
|
|
return
|
|
}
|
|
timing.response(ctx, f)
|
|
}
|
|
}()
|
|
for {
|
|
f, err := read(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if p.AccessMode == "readonly" && !readonlyAllowed(f) {
|
|
if event != nil {
|
|
event(model.EventReadOnlyRejected, "rabbitmq_readonly")
|
|
}
|
|
if err = model.AuditCommand(ctx, "readonly.reject", fmt.Sprintf("channel=%d", f.channel), "denied"); err != nil {
|
|
return err
|
|
}
|
|
// Close the stream so queued publish content cannot become orphan frames.
|
|
message := "gateway readonly policy: only passive inspection is allowed"
|
|
body := binary.BigEndian.AppendUint16([]byte{0, 10, 0, 50}, 403)
|
|
body = append(body, byte(len(message)))
|
|
body = append(body, message...)
|
|
if f.kind == 1 && len(f.body) >= 4 {
|
|
body = append(body, f.body[:4]...)
|
|
} else {
|
|
body = append(body, 0, 0, 0, 0)
|
|
}
|
|
if err = write(out, frame{kind: 1, body: body}); err != nil {
|
|
return err
|
|
}
|
|
return errors.New("AMQP gateway readonly policy rejected client frame")
|
|
}
|
|
if f.kind == 1 {
|
|
if len(f.body) < 4 {
|
|
return errors.New("invalid AMQP method")
|
|
}
|
|
class, methodID := binary.BigEndian.Uint16(f.body), binary.BigEndian.Uint16(f.body[2:])
|
|
kind := fmt.Sprintf("method.%d.%d", class, methodID)
|
|
names := map[string]string{"method.50.10": "queue.declare", "method.50.20": "queue.bind", "method.50.40": "queue.delete", "method.40.10": "exchange.declare", "method.60.40": "basic.publish", "method.60.20": "basic.consume", "method.60.70": "basic.get", "method.60.80": "basic.ack", "method.60.90": "basic.reject", "method.60.120": "basic.nack", "method.85.10": "confirm.select", "method.90.10": "tx.select", "method.90.20": "tx.commit", "method.90.30": "tx.rollback"}
|
|
if name := names[kind]; name != "" {
|
|
kind = name
|
|
}
|
|
if class == 10 && (methodID == 11 || methodID == 70) {
|
|
return errors.New("AMQP authentication changes are unsupported")
|
|
}
|
|
detail := fmt.Sprintf("channel=%d method=%d.%d bytes=%d", f.channel, class, methodID, len(f.body))
|
|
if err = model.AuditCommand(ctx, kind, detail, "allowed"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
timing.request(f)
|
|
if err = write(up, f); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|