Support multiple service port forwardings

README.md

@@ -16,7 +16,7 @@ reference is placed in the `MikroTikRouter` resource. Every router must
 explicitly list the namespaces permitted to create Internet-facing rules; an
 empty allowlist permits none.
 
-One annotated Service creates one rule in this first revision:
+For a Service with one public port, the compact annotations create one rule:
 
 ```yaml
 metadata:
@@ -29,8 +29,34 @@ metadata:
 
 `service-port` selects a Service port by name or number and is optional when the
 Service has exactly one port. `dst-port` is the public port and defaults to the
-selected Service port. Protocol is inherited from that port. The generated
-RouterOS properties are:
+selected Service port. For multiple public ports, use the `forwardings`
+annotation with uniquely named entries:
+
+```yaml
+metadata:
+  annotations:
+    mikrotik.brunner.ninja/router: newgw
+    mikrotik.brunner.ninja/forwardings: >-
+      [
+        {"name":"http","servicePort":"web","dstAddressList":"WANs"},
+        {"name":"https","servicePort":"websecure","dstAddressList":"WANs"}
+      ]
+```
+
+Each entry accepts `servicePort`, `dstPort`, `dstAddress`, `dstAddressList`,
+`inInterface`, `inInterfaceList`, `srcAddress`, and `srcAddressList`. Protocol
+is inherited from the selected Service port. A contiguous range can be mapped
+without creating one RouterOS rule per port when every port in the range exists
+on the Service and uses the same protocol:
+
+```yaml
+mikrotik.brunner.ninja/forwardings: >-
+  [{"name":"relay","servicePortRange":"49160-49200","dstAddressList":"WANs"}]
+```
+
+`dstPort` defaults to `servicePort` or `servicePortRange`; if overridden, its
+range must contain the same number of ports. The generated RouterOS properties
+are:
 
 | Source | RouterOS property |
 | --- | --- |
@@ -48,11 +74,13 @@ RouterOS properties are:
 RouterOS calls the incoming/WAN matcher `in-interface`, not `src-interface`.
 `src-address` refers to the remote client's address.
 
-Managed rules have comments of the form
-`k8s-mikrotik/<service-namespace>/<service-name>`. The controller never modifies
-rules without that prefix. It creates and updates desired rules, deletes stale
-owned rules, removes duplicate owned rules, and preserves an existing rule if
-the corresponding Service temporarily has no LoadBalancer IP.
+Single managed rules have comments of the form
+`k8s-mikrotik/<service-namespace>/<service-name>`. Named forwardings append
+`/<forwarding-name>`. The controller never modifies rules without the
+`k8s-mikrotik/` prefix. It creates and updates desired rules, deletes stale
+owned rules, removes duplicate owned rules, and preserves existing rules if the
+corresponding Service temporarily has no LoadBalancer IP or invalid annotation
+JSON.
 
 Deleting a `MikroTikRouter` removes all of its managed rules before the CR is
 deleted. If RouterOS cannot be reached, the cleanup finalizer deliberately keeps
@@ -61,11 +89,12 @@ the CR in `Terminating` rather than silently leaving public ports open.
 ## RouterOS preparation
 
 The implementation uses the RouterOS v7 REST API. Enable `www-ssl`, configure a
-certificate, and create a dedicated user rather than using `admin`. The minimum
-group policies needed are `read`, `write`, and `rest-api`:
+certificate, and create a dedicated user rather than using `admin`. RouterOS
+7.24 on `newgw` also requires the `api` policy for REST operations, so the
+verified group policies are `read`, `write`, `api`, and `rest-api`:
 
 ```routeros
-/user/group/add name=kubernetes-nat policy=read,write,rest-api
+/user/group/add name=kubernetes-nat policy=read,write,api,rest-api
 /user/add name=kubernetes-nat group=kubernetes-nat password=<strong-password> address=<operator-egress-cidr>
 /ip/service/enable www-ssl
 ```
@@ -179,8 +208,6 @@ tls:
 ## Current limits and next revisions
 
 - RouterOS destination NAT is IPv4-only; IPv6 LoadBalancer addresses are ignored.
-- One Service currently produces one NAT rule. Split Services by exposed port,
-  or add a typed multi-forwarding API in the next revision.
 - There is no admission webhook yet. Validation errors are reported in the
   router's `Ready` condition while the previous rule is preserved.
 - Metrics use controller-runtime's standard endpoint; domain-specific counters

chart/mikrotik-operator/crds/mikrotikrouters.yaml

@@ -77,10 +77,10 @@ spec:
                           minLength: 1
                 allowedNamespaces:
                   type: array
+                  x-kubernetes-list-type: set
                   items:
                     type: string
                     minLength: 1
-                  uniqueItems: true
             status:
               type: object
               properties:

chart/mikrotik-operator/templates/rbac.yaml

@@ -49,6 +49,9 @@ rules:
   - apiGroups: ["coordination.k8s.io"]
     resources: ["leases"]
     verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
+  - apiGroups: [""]
+    resources: ["events"]
+    verbs: ["create", "patch", "update"]
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding

cmd/manager/main.go

@@ -8,6 +8,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
@@ -34,7 +35,10 @@ func main() {
 	must(mikrotikv1alpha1.AddToScheme(scheme))
 
 	manager, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
-		Scheme:                 scheme,
+		Scheme: scheme,
+		Client: client.Options{Cache: &client.CacheOptions{
+			DisableFor: []client.Object{&corev1.Secret{}},
+		}},
 		Metrics:                metricsserver.Options{BindAddress: metricsAddress},
 		HealthProbeBindAddress: healthAddress,
 		LeaderElection:         leaderElection,

config/crd/mikrotikrouters.yaml

@@ -77,10 +77,10 @@ spec:
                           minLength: 1
                 allowedNamespaces:
                   type: array
+                  x-kubernetes-list-type: set
                   items:
                     type: string
                     minLength: 1
-                  uniqueItems: true
             status:
               type: object
               properties:

config/rbac/operator.yaml

@@ -53,6 +53,9 @@ rules:
   - apiGroups: ["coordination.k8s.io"]
     resources: ["leases"]
     verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
+  - apiGroups: [""]
+    resources: ["events"]
+    verbs: ["create", "patch", "update"]
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding

examples/router.yaml

@@ -3,12 +3,9 @@ kind: MikroTikRouter
 metadata:
   name: newgw
 spec:
-  endpoint: https://newgw.localdomain
+  endpoint: http://192.168.0.254
   credentialsSecretRef:
     namespace: mikrotik-operator-system
     name: newgw-credentials
-  tls:
-    # Use a caSecretRef in production. This is convenient only for first contact.
-    insecureSkipVerify: true
   allowedNamespaces:
-    - factorio
+    - "*"

examples/service.yaml

@@ -6,9 +6,8 @@ metadata:
   annotations:
     metallb.io/loadBalancerIPs: "192.168.0.17"
     mikrotik.brunner.ninja/router: newgw
-    mikrotik.brunner.ninja/service-port: game-udp
-    mikrotik.brunner.ninja/dst-port: "34197"
-    mikrotik.brunner.ninja/in-interface-list: WAN
+    mikrotik.brunner.ninja/forwardings: >-
+      [{"name":"game","servicePort":"game-udp","dstAddressList":"WANs"}]
     # Optional matchers:
     # mikrotik.brunner.ninja/dst-address: 203.0.113.10
     # mikrotik.brunner.ninja/dst-address-list: public-addresses

internal/controller/service_rules.go

@@ -1,12 +1,16 @@
 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"
@@ -24,8 +28,22 @@ const (
 	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
@@ -40,10 +58,30 @@ func DesiredRules(router *mikrotikv1alpha1.MikroTikRouter, services []corev1.Ser
 		if service.Annotations[AnnotationRouter] != router.Name || !namespaceAllowed(router.Spec.AllowedNamespaces, service.Namespace) {
 			continue
 		}
-		comment := routeros.ManagedComment(service.Namespace, service.Name)
-		rule, err := ruleForService(service, comment)
+		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[comment] = struct{}{}
+			preserve[baseComment] = struct{}{}
 			issues = append(issues, RuleIssue{Service: service.Namespace + "/" + service.Name, Message: err.Error()})
 			continue
 		}
@@ -52,6 +90,47 @@ func DesiredRules(router *mikrotikv1alpha1.MikroTikRouter, services []corev1.Ser
 	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 {
@@ -103,6 +182,112 @@ func ruleForService(service *corev1.Service, comment string) (routeros.Rule, err
 	}, 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 == "" {

internal/controller/service_rules_test.go

@@ -86,3 +86,68 @@ func TestDesiredRulesDoesNotHonorUnlistedNamespace(t *testing.T) {
 		t.Fatalf("unlisted namespace affected reconciliation: rules=%#v preserve=%#v issues=%#v", rules, preserve, issues)
 	}
 }
+
+func TestDesiredRulesSupportsNamedForwardingsAndRanges(t *testing.T) {
+	router := &mikrotikv1alpha1.MikroTikRouter{
+		ObjectMeta: metav1.ObjectMeta{Name: "edge"},
+		Spec:       mikrotikv1alpha1.MikroTikRouterSpec{AllowedNamespaces: []string{"*"}},
+	}
+	service := corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "coturn", Name: "coturn",
+			Annotations: map[string]string{
+				AnnotationRouter: "edge",
+				AnnotationForwardings: `[
+					{"name":"turn","servicePort":"turn","dstAddressList":"WANs"},
+					{"name":"relay","servicePortRange":"49160-49162","dstAddressList":"WANs"}
+				]`,
+			},
+		},
+		Spec: corev1.ServiceSpec{
+			Type: corev1.ServiceTypeLoadBalancer,
+			Ports: []corev1.ServicePort{
+				{Name: "turn", Port: 3478, Protocol: corev1.ProtocolUDP},
+				{Name: "relay-49160", Port: 49160, Protocol: corev1.ProtocolUDP},
+				{Name: "relay-49161", Port: 49161, Protocol: corev1.ProtocolUDP},
+				{Name: "relay-49162", Port: 49162, Protocol: corev1.ProtocolUDP},
+			},
+		},
+		Status: corev1.ServiceStatus{LoadBalancer: corev1.LoadBalancerStatus{Ingress: []corev1.LoadBalancerIngress{{IP: "192.168.0.19"}}}},
+	}
+
+	rules, preserve, issues := DesiredRules(router, []corev1.Service{service})
+	if len(issues) != 0 || len(preserve) != 0 {
+		t.Fatalf("issues=%#v preserve=%#v", issues, preserve)
+	}
+	if len(rules) != 2 {
+		t.Fatalf("got %d rules, want 2", len(rules))
+	}
+	want := []routeros.Rule{
+		{Comment: routeros.ManagedForwardingComment("coturn", "coturn", "turn"), Protocol: "udp", DestinationPort: "3478", DestinationList: "WANs", ToAddress: "192.168.0.19", ToPort: "3478"},
+		{Comment: routeros.ManagedForwardingComment("coturn", "coturn", "relay"), Protocol: "udp", DestinationPort: "49160-49162", DestinationList: "WANs", ToAddress: "192.168.0.19", ToPort: "49160-49162"},
+	}
+	for i := range want {
+		if rules[i] != want[i] {
+			t.Fatalf("rule[%d] = %#v, want %#v", i, rules[i], want[i])
+		}
+	}
+}
+
+func TestDesiredRulesPreservesAllNamedRulesForInvalidJSON(t *testing.T) {
+	router := &mikrotikv1alpha1.MikroTikRouter{
+		ObjectMeta: metav1.ObjectMeta{Name: "edge"},
+		Spec:       mikrotikv1alpha1.MikroTikRouterSpec{AllowedNamespaces: []string{"*"}},
+	}
+	service := corev1.Service{ObjectMeta: metav1.ObjectMeta{
+		Namespace: "default", Name: "traefik", Annotations: map[string]string{
+			AnnotationRouter: "edge", AnnotationForwardings: `[{broken]`,
+		},
+	}}
+	rules, preserve, issues := DesiredRules(router, []corev1.Service{service})
+	if len(rules) != 0 || len(issues) != 1 {
+		t.Fatalf("rules=%#v issues=%#v", rules, issues)
+	}
+	if _, ok := preserve[routeros.ManagedComment("default", "traefik")+"/"]; !ok {
+		t.Fatalf("named forwarding prefix was not preserved: %#v", preserve)
+	}
+}

internal/routeros/client.go

@@ -159,7 +159,7 @@ func (c *Client) Sync(ctx context.Context, desired []Rule, preserve map[string]s
 		if _, wanted := desiredComments[comment]; wanted {
 			continue
 		}
-		if _, keep := preserve[comment]; keep {
+		if commentPreserved(comment, preserve) {
 			managedCount++
 			continue
 		}
@@ -172,6 +172,18 @@ func (c *Client) Sync(ctx context.Context, desired []Rule, preserve map[string]s
 	return managedCount, nil
 }
 
+func commentPreserved(comment string, preserve map[string]struct{}) bool {
+	if _, ok := preserve[comment]; ok {
+		return true
+	}
+	for prefix := range preserve {
+		if strings.HasSuffix(prefix, "/") && strings.HasPrefix(comment, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
 func sameManagedFields(current, desired map[string]string) bool {
 	for _, key := range managedFields {
 		if current[key] != desired[key] {
@@ -245,3 +257,7 @@ func (c *Client) do(ctx context.Context, method string, path []string, input, ou
 func ManagedComment(namespace, name string) string {
 	return managedCommentPrefix + namespace + "/" + name
 }
+
+func ManagedForwardingComment(namespace, service, forwarding string) string {
+	return ManagedComment(namespace, service) + "/" + forwarding
+}

internal/routeros/client_test.go

@@ -74,6 +74,16 @@ func TestSyncOwnsOnlyPrefixedRules(t *testing.T) {
 	}
 }
 
+func TestCommentPreservedByServicePrefix(t *testing.T) {
+	preserve := map[string]struct{}{"k8s-mikrotik/default/traefik/": {}}
+	if !commentPreserved("k8s-mikrotik/default/traefik/http", preserve) {
+		t.Fatal("named rule was not preserved")
+	}
+	if commentPreserved("k8s-mikrotik/default/other/http", preserve) {
+		t.Fatal("unrelated rule was preserved")
+	}
+}
+
 type roundTripFunc func(*http.Request) (*http.Response, error)
 
 func (function roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {