306 lines
11 KiB
Go
306 lines
11 KiB
Go
package controller
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/netip"
|
|
"strconv"
|
|
"strings"
|
|
|
|
corev1 "k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/util/validation"
|
|
|
|
mikrotikv1alpha1 "code.brunner.ninja/feedc0de/mikrotik-operator/api/v1alpha1"
|
|
"code.brunner.ninja/feedc0de/mikrotik-operator/internal/routeros"
|
|
)
|
|
|
|
const AnnotationPrefix = "mikrotik.brunner.ninja/"
|
|
|
|
const (
|
|
AnnotationRouter = AnnotationPrefix + "router"
|
|
AnnotationServicePort = AnnotationPrefix + "service-port"
|
|
AnnotationDestinationPort = AnnotationPrefix + "dst-port"
|
|
AnnotationDestinationIP = AnnotationPrefix + "dst-address"
|
|
AnnotationDestinationList = AnnotationPrefix + "dst-address-list"
|
|
AnnotationInInterface = AnnotationPrefix + "in-interface"
|
|
AnnotationInInterfaceList = AnnotationPrefix + "in-interface-list"
|
|
AnnotationSourceAddress = AnnotationPrefix + "src-address"
|
|
AnnotationSourceList = AnnotationPrefix + "src-address-list"
|
|
AnnotationForwardings = AnnotationPrefix + "forwardings"
|
|
)
|
|
|
|
type forwardingSpec struct {
|
|
Name string `json:"name"`
|
|
ServicePort string `json:"servicePort,omitempty"`
|
|
ServicePortRange string `json:"servicePortRange,omitempty"`
|
|
DestinationPort string `json:"dstPort,omitempty"`
|
|
DestinationIP string `json:"dstAddress,omitempty"`
|
|
DestinationList string `json:"dstAddressList,omitempty"`
|
|
InInterface string `json:"inInterface,omitempty"`
|
|
InInterfaceList string `json:"inInterfaceList,omitempty"`
|
|
SourceAddress string `json:"srcAddress,omitempty"`
|
|
SourceList string `json:"srcAddressList,omitempty"`
|
|
}
|
|
|
|
type RuleIssue struct {
|
|
Service string
|
|
Message string
|
|
}
|
|
|
|
func DesiredRules(router *mikrotikv1alpha1.MikroTikRouter, services []corev1.Service) ([]routeros.Rule, map[string]struct{}, []RuleIssue) {
|
|
desired := make([]routeros.Rule, 0)
|
|
preserve := make(map[string]struct{})
|
|
issues := make([]RuleIssue, 0)
|
|
for i := range services {
|
|
service := &services[i]
|
|
if service.Annotations[AnnotationRouter] != router.Name || !namespaceAllowed(router.Spec.AllowedNamespaces, service.Namespace) {
|
|
continue
|
|
}
|
|
baseComment := routeros.ManagedComment(service.Namespace, service.Name)
|
|
if encoded := service.Annotations[AnnotationForwardings]; encoded != "" {
|
|
forwardings, err := parseForwardings(encoded)
|
|
if err != nil {
|
|
preserve[baseComment] = struct{}{}
|
|
preserve[baseComment+"/"] = struct{}{}
|
|
issues = append(issues, RuleIssue{Service: service.Namespace + "/" + service.Name, Message: err.Error()})
|
|
continue
|
|
}
|
|
for _, forwarding := range forwardings {
|
|
comment := routeros.ManagedForwardingComment(service.Namespace, service.Name, forwarding.Name)
|
|
rule, err := ruleForForwarding(service, comment, forwarding)
|
|
if err != nil {
|
|
preserve[comment] = struct{}{}
|
|
issues = append(issues, RuleIssue{Service: service.Namespace + "/" + service.Name + "/" + forwarding.Name, Message: err.Error()})
|
|
continue
|
|
}
|
|
desired = append(desired, rule)
|
|
}
|
|
continue
|
|
}
|
|
rule, err := ruleForService(service, baseComment)
|
|
if err != nil {
|
|
preserve[baseComment] = struct{}{}
|
|
issues = append(issues, RuleIssue{Service: service.Namespace + "/" + service.Name, Message: err.Error()})
|
|
continue
|
|
}
|
|
desired = append(desired, rule)
|
|
}
|
|
return desired, preserve, issues
|
|
}
|
|
|
|
func parseForwardings(encoded string) ([]forwardingSpec, error) {
|
|
decoder := json.NewDecoder(bytes.NewBufferString(encoded))
|
|
decoder.DisallowUnknownFields()
|
|
var forwardings []forwardingSpec
|
|
if err := decoder.Decode(&forwardings); err != nil {
|
|
return nil, fmt.Errorf("%s must be a JSON array: %w", AnnotationForwardings, err)
|
|
}
|
|
if err := ensureJSONEOF(decoder); err != nil {
|
|
return nil, fmt.Errorf("%s contains trailing data", AnnotationForwardings)
|
|
}
|
|
if len(forwardings) == 0 {
|
|
return nil, fmt.Errorf("%s must contain at least one forwarding", AnnotationForwardings)
|
|
}
|
|
seen := make(map[string]struct{}, len(forwardings))
|
|
for _, forwarding := range forwardings {
|
|
if problems := validation.IsDNS1123Label(forwarding.Name); len(problems) != 0 {
|
|
return nil, fmt.Errorf("forwarding name %q must be a DNS label: %s", forwarding.Name, strings.Join(problems, ", "))
|
|
}
|
|
if _, exists := seen[forwarding.Name]; exists {
|
|
return nil, fmt.Errorf("forwarding name %q is duplicated", forwarding.Name)
|
|
}
|
|
seen[forwarding.Name] = struct{}{}
|
|
if forwarding.ServicePort != "" && forwarding.ServicePortRange != "" {
|
|
return nil, fmt.Errorf("forwarding %q must not set both servicePort and servicePortRange", forwarding.Name)
|
|
}
|
|
}
|
|
return forwardings, nil
|
|
}
|
|
|
|
func ensureJSONEOF(decoder *json.Decoder) error {
|
|
var extra any
|
|
err := decoder.Decode(&extra)
|
|
if err == io.EOF {
|
|
return nil
|
|
}
|
|
if err == nil {
|
|
return fmt.Errorf("unexpected additional JSON value")
|
|
}
|
|
return err
|
|
}
|
|
|
|
func namespaceAllowed(allowed []string, namespace string) bool {
|
|
for _, item := range allowed {
|
|
if item == "*" || item == namespace {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func ruleForService(service *corev1.Service, comment string) (routeros.Rule, error) {
|
|
if service.Spec.Type != corev1.ServiceTypeLoadBalancer {
|
|
return routeros.Rule{}, fmt.Errorf("Service must have type LoadBalancer")
|
|
}
|
|
port, err := selectServicePort(service)
|
|
if err != nil {
|
|
return routeros.Rule{}, err
|
|
}
|
|
toAddress := ""
|
|
for _, ingress := range service.Status.LoadBalancer.Ingress {
|
|
address, parseErr := netip.ParseAddr(ingress.IP)
|
|
if parseErr == nil && address.Is4() {
|
|
toAddress = address.String()
|
|
break
|
|
}
|
|
}
|
|
if toAddress == "" {
|
|
return routeros.Rule{}, fmt.Errorf("waiting for an IPv4 address in status.loadBalancer.ingress")
|
|
}
|
|
|
|
destinationPort := service.Annotations[AnnotationDestinationPort]
|
|
if destinationPort == "" {
|
|
destinationPort = strconv.Itoa(int(port.Port))
|
|
}
|
|
parsedDestinationPort, err := strconv.ParseUint(destinationPort, 10, 16)
|
|
if err != nil || parsedDestinationPort == 0 {
|
|
return routeros.Rule{}, fmt.Errorf("%s must be a port from 1 through 65535", AnnotationDestinationPort)
|
|
}
|
|
|
|
return routeros.Rule{
|
|
Comment: comment, Protocol: strings.ToLower(string(port.Protocol)),
|
|
DestinationPort: destinationPort,
|
|
DestinationIP: service.Annotations[AnnotationDestinationIP],
|
|
DestinationList: service.Annotations[AnnotationDestinationList],
|
|
InInterface: service.Annotations[AnnotationInInterface],
|
|
InInterfaceList: service.Annotations[AnnotationInInterfaceList],
|
|
SourceAddress: service.Annotations[AnnotationSourceAddress],
|
|
SourceList: service.Annotations[AnnotationSourceList],
|
|
ToAddress: toAddress, ToPort: strconv.Itoa(int(port.Port)),
|
|
}, nil
|
|
}
|
|
|
|
func ruleForForwarding(service *corev1.Service, comment string, forwarding forwardingSpec) (routeros.Rule, error) {
|
|
if service.Spec.Type != corev1.ServiceTypeLoadBalancer {
|
|
return routeros.Rule{}, fmt.Errorf("Service must have type LoadBalancer")
|
|
}
|
|
protocol, toPort, err := selectForwardingPorts(service, forwarding)
|
|
if err != nil {
|
|
return routeros.Rule{}, err
|
|
}
|
|
toAddress, err := loadBalancerIPv4(service)
|
|
if err != nil {
|
|
return routeros.Rule{}, err
|
|
}
|
|
destinationPort := forwarding.DestinationPort
|
|
if destinationPort == "" {
|
|
destinationPort = toPort
|
|
}
|
|
destinationStart, destinationEnd, err := parsePortOrRange(destinationPort)
|
|
if err != nil {
|
|
return routeros.Rule{}, fmt.Errorf("forwarding %q dstPort: %w", forwarding.Name, err)
|
|
}
|
|
toStart, toEnd, err := parsePortOrRange(toPort)
|
|
if err != nil {
|
|
return routeros.Rule{}, fmt.Errorf("forwarding %q internal error: %w", forwarding.Name, err)
|
|
}
|
|
if destinationEnd-destinationStart != toEnd-toStart {
|
|
return routeros.Rule{}, fmt.Errorf("forwarding %q dstPort and service port range must contain the same number of ports", forwarding.Name)
|
|
}
|
|
return routeros.Rule{
|
|
Comment: comment, Protocol: protocol, DestinationPort: destinationPort,
|
|
DestinationIP: forwarding.DestinationIP, DestinationList: forwarding.DestinationList,
|
|
InInterface: forwarding.InInterface, InInterfaceList: forwarding.InInterfaceList,
|
|
SourceAddress: forwarding.SourceAddress, SourceList: forwarding.SourceList,
|
|
ToAddress: toAddress, ToPort: toPort,
|
|
}, nil
|
|
}
|
|
|
|
func selectForwardingPorts(service *corev1.Service, forwarding forwardingSpec) (string, string, error) {
|
|
if forwarding.ServicePortRange == "" {
|
|
copy := service.DeepCopy()
|
|
copy.Annotations = make(map[string]string)
|
|
if forwarding.ServicePort != "" {
|
|
copy.Annotations[AnnotationServicePort] = forwarding.ServicePort
|
|
}
|
|
port, err := selectServicePort(copy)
|
|
if err != nil {
|
|
return "", "", fmt.Errorf("forwarding %q: %w", forwarding.Name, err)
|
|
}
|
|
return strings.ToLower(string(port.Protocol)), strconv.Itoa(int(port.Port)), nil
|
|
}
|
|
|
|
start, end, err := parsePortOrRange(forwarding.ServicePortRange)
|
|
if err != nil || start == end {
|
|
return "", "", fmt.Errorf("forwarding %q servicePortRange must be a range such as 49160-49200", forwarding.Name)
|
|
}
|
|
protocol := corev1.Protocol("")
|
|
for number := start; number <= end; number++ {
|
|
found := false
|
|
for _, port := range service.Spec.Ports {
|
|
if int(port.Port) != number {
|
|
continue
|
|
}
|
|
found = true
|
|
if protocol == "" {
|
|
protocol = port.Protocol
|
|
} else if protocol != port.Protocol {
|
|
return "", "", fmt.Errorf("forwarding %q servicePortRange contains mixed protocols", forwarding.Name)
|
|
}
|
|
break
|
|
}
|
|
if !found {
|
|
return "", "", fmt.Errorf("forwarding %q servicePortRange includes port %d, which the Service does not expose", forwarding.Name, number)
|
|
}
|
|
}
|
|
return strings.ToLower(string(protocol)), forwarding.ServicePortRange, nil
|
|
}
|
|
|
|
func parsePortOrRange(value string) (int, int, error) {
|
|
parts := strings.Split(value, "-")
|
|
if len(parts) < 1 || len(parts) > 2 {
|
|
return 0, 0, fmt.Errorf("must be a port or ascending port range from 1 through 65535")
|
|
}
|
|
start, err := strconv.ParseUint(parts[0], 10, 16)
|
|
if err != nil || start == 0 {
|
|
return 0, 0, fmt.Errorf("must be a port or ascending port range from 1 through 65535")
|
|
}
|
|
end := start
|
|
if len(parts) == 2 {
|
|
parsedEnd, parseErr := strconv.ParseUint(parts[1], 10, 16)
|
|
if parseErr != nil || parsedEnd == 0 || parsedEnd < start {
|
|
return 0, 0, fmt.Errorf("must be a port or ascending port range from 1 through 65535")
|
|
}
|
|
end = parsedEnd
|
|
}
|
|
return int(start), int(end), nil
|
|
}
|
|
|
|
func loadBalancerIPv4(service *corev1.Service) (string, error) {
|
|
for _, ingress := range service.Status.LoadBalancer.Ingress {
|
|
address, err := netip.ParseAddr(ingress.IP)
|
|
if err == nil && address.Is4() {
|
|
return address.String(), nil
|
|
}
|
|
}
|
|
return "", fmt.Errorf("waiting for an IPv4 address in status.loadBalancer.ingress")
|
|
}
|
|
|
|
func selectServicePort(service *corev1.Service) (corev1.ServicePort, error) {
|
|
selector := service.Annotations[AnnotationServicePort]
|
|
if selector == "" {
|
|
if len(service.Spec.Ports) != 1 {
|
|
return corev1.ServicePort{}, fmt.Errorf("%s is required when the Service has more than one port", AnnotationServicePort)
|
|
}
|
|
return service.Spec.Ports[0], nil
|
|
}
|
|
for _, port := range service.Spec.Ports {
|
|
if port.Name == selector || strconv.Itoa(int(port.Port)) == selector {
|
|
return port, nil
|
|
}
|
|
}
|
|
return corev1.ServicePort{}, fmt.Errorf("%s %q does not match a Service port name or number", AnnotationServicePort, selector)
|
|
}
|