Skip to content

Commit

Permalink
L7 Visibility Annotations
Browse files Browse the repository at this point in the history 
Translating VisibilityPolicy into PortNetwrokPolicy to enable
L7 visibility for pod annotations.

Signed-off-by: Thales Paiva <thales@accuknox.com>
  • Loading branch information
@trvll
trvll committed Oct 26, 2021
1 parent 5a6a51b commit 54b81ae
Show file tree
Hide file tree
Showing 8 changed files with 122 additions and 24 deletions.
2 changes: 1 addition & 1 deletion pkg/endpoint/policy.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ func (e *Endpoint) updateNetworkPolicy(proxyWaitGroup *completion.WaitGroup) (re
}

// Publish the updated policy to L7 proxies.
return e.proxy.UpdateNetworkPolicy(e, e.desiredPolicy.L4Policy, e.desiredPolicy.IngressPolicyEnabled, e.desiredPolicy.EgressPolicyEnabled, proxyWaitGroup)
return e.proxy.UpdateNetworkPolicy(e, e.visibilityPolicy, e.desiredPolicy.L4Policy, e.desiredPolicy.IngressPolicyEnabled, e.desiredPolicy.EgressPolicyEnabled, proxyWaitGroup)
}

func (e *Endpoint) useCurrentNetworkPolicy(proxyWaitGroup *completion.WaitGroup) {
Expand Down
4 changes: 2 additions & 2 deletions pkg/endpoint/proxy.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import (
type EndpointProxy interface {
CreateOrUpdateRedirect(l4 policy.ProxyPolicy, id string, localEndpoint logger.EndpointUpdater, wg *completion.WaitGroup) (proxyPort uint16, err error, finalizeFunc revert.FinalizeFunc, revertFunc revert.RevertFunc)
RemoveRedirect(id string, wg *completion.WaitGroup) (error, revert.FinalizeFunc, revert.RevertFunc)
UpdateNetworkPolicy(ep logger.EndpointUpdater, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error)
UpdateNetworkPolicy(ep logger.EndpointUpdater, vis *policy.VisibilityPolicy, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error)
UseCurrentNetworkPolicy(ep logger.EndpointUpdater, policy *policy.L4Policy, wg *completion.WaitGroup)
RemoveNetworkPolicy(ep logger.EndpointInfoSource)
}
Expand Down Expand Up @@ -53,7 +53,7 @@ func (f *FakeEndpointProxy) RemoveRedirect(id string, wg *completion.WaitGroup)
}

// UpdateNetworkPolicy does nothing.
func (f *FakeEndpointProxy) UpdateNetworkPolicy(ep logger.EndpointUpdater, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error) {
func (f *FakeEndpointProxy) UpdateNetworkPolicy(ep logger.EndpointUpdater, vis *policy.VisibilityPolicy, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error) {
return nil, nil
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/endpoint/redirect_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ func (r *RedirectSuiteProxy) RemoveRedirect(id string, wg *completion.WaitGroup)
}

// UpdateNetworkPolicy does nothing.
func (r *RedirectSuiteProxy) UpdateNetworkPolicy(ep logger.EndpointUpdater, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error) {
func (r *RedirectSuiteProxy) UpdateNetworkPolicy(ep logger.EndpointUpdater, vis *policy.VisibilityPolicy, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error) {
return nil, nil
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/endpointmanager/manager_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ func (s *EndpointManagerSuite) RemoveProxyRedirect(e regeneration.EndpointInfoSo
return nil, nil, nil
}

func (s *EndpointManagerSuite) UpdateNetworkPolicy(e regeneration.EndpointUpdater, policy *policy.L4Policy,
func (s *EndpointManagerSuite) UpdateNetworkPolicy(e regeneration.EndpointUpdater, vis *policy.VisibilityPolicy, policy *policy.L4Policy,
proxyWaitGroup *completion.WaitGroup) (error, revert.RevertFunc) {
return nil, nil
}
Expand Down
66 changes: 60 additions & 6 deletions pkg/envoy/server.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"github.com/cilium/cilium/pkg/policy/api"
"github.com/cilium/cilium/pkg/policy/api/kafka"
"github.com/cilium/cilium/pkg/proxy/logger"
"github.com/cilium/cilium/pkg/u8proto"

cilium "github.com/cilium/proxy/go/cilium/api"
envoy_config_bootstrap "github.com/cilium/proxy/go/envoy/config/bootstrap/v3"
Expand Down Expand Up @@ -1334,7 +1335,7 @@ func getDirectionNetworkPolicy(ep logger.EndpointUpdater, l4Policy policy.L4Poli
}

// getNetworkPolicy converts a network policy into a cilium.NetworkPolicy.
func getNetworkPolicy(ep logger.EndpointUpdater, name string, policy *policy.L4Policy,
func getNetworkPolicy(ep logger.EndpointUpdater, vis *policy.VisibilityPolicy, name string, l4Policy *policy.L4Policy,
ingressPolicyEnforced, egressPolicyEnforced bool) *cilium.NetworkPolicy {
p := &cilium.NetworkPolicy{
Name: name,
Expand All @@ -1343,9 +1344,62 @@ func getNetworkPolicy(ep logger.EndpointUpdater, name string, policy *policy.L4P
}

// If no policy, deny all traffic. Otherwise, convert the policies for ingress and egress.
if policy != nil {
p.IngressPerPortPolicies = getDirectionNetworkPolicy(ep, policy.Ingress, ingressPolicyEnforced)
p.EgressPerPortPolicies = getDirectionNetworkPolicy(ep, policy.Egress, egressPolicyEnforced)
if l4Policy != nil {
if vis != nil && !(ingressPolicyEnforced || egressPolicyEnforced) {
if vis.Ingress != nil {
PerPortPolicies := make([]*cilium.PortNetworkPolicy, 0, len(vis.Ingress))
for _, visMeta := range vis.Ingress {
// we only setup this for proxylib parsers
if visMeta.Parser != policy.ParserTypeHTTP && visMeta.Parser != policy.ParserTypeDNS {
rules := []*cilium.PortNetworkPolicyRule{
{
L7Proto: visMeta.Parser.String(),
},
}
if visMeta.Proto != u8proto.TCP {
PerPortPolicies = allowAllPortNetworkPolicy
} else {
protocol := envoy_config_core.SocketAddress_TCP

PerPortPolicies = append(PerPortPolicies, &cilium.PortNetworkPolicy{
Port: uint32(visMeta.Port),
Protocol: protocol,
Rules: rules,
})
}
}
}
p.IngressPerPortPolicies = SortPortNetworkPolicies(PerPortPolicies)
}
if vis.Egress != nil {
PerPortPolicies := make([]*cilium.PortNetworkPolicy, 0, len(vis.Egress))
for _, visMeta := range vis.Egress {
// we only setup this for proxylib parsers
if visMeta.Parser != policy.ParserTypeHTTP && visMeta.Parser != policy.ParserTypeDNS {
rules := []*cilium.PortNetworkPolicyRule{
{
L7Proto: visMeta.Parser.String(),
},
}
if visMeta.Proto != u8proto.TCP {
PerPortPolicies = allowAllPortNetworkPolicy
} else {
protocol := envoy_config_core.SocketAddress_TCP

PerPortPolicies = append(PerPortPolicies, &cilium.PortNetworkPolicy{
Port: uint32(visMeta.Port),
Protocol: protocol,
Rules: rules,
})
}
}
}
p.EgressPerPortPolicies = SortPortNetworkPolicies(PerPortPolicies)
}
} else {
p.IngressPerPortPolicies = getDirectionNetworkPolicy(ep, l4Policy.Ingress, ingressPolicyEnforced)
p.EgressPerPortPolicies = getDirectionNetworkPolicy(ep, l4Policy.Egress, egressPolicyEnforced)
}
}

return p
Expand Down Expand Up @@ -1384,7 +1438,7 @@ func getNodeIDs(ep logger.EndpointUpdater, policy *policy.L4Policy) []string {
// to L7 proxies.
// When the proxy acknowledges the network policy update, it will result in
// a subsequent call to the endpoint's OnProxyPolicyUpdate() function.
func (s *XDSServer) UpdateNetworkPolicy(ep logger.EndpointUpdater, policy *policy.L4Policy,
func (s *XDSServer) UpdateNetworkPolicy(ep logger.EndpointUpdater, vis *policy.VisibilityPolicy, policy *policy.L4Policy,
ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error) {
s.mutex.Lock()
defer s.mutex.Unlock()
Expand All @@ -1399,7 +1453,7 @@ func (s *XDSServer) UpdateNetworkPolicy(ep logger.EndpointUpdater, policy *polic
if ip == "" {
continue
}
networkPolicy := getNetworkPolicy(ep, ip, policy, ingressPolicyEnforced, egressPolicyEnforced)
networkPolicy := getNetworkPolicy(ep, vis, ip, policy, ingressPolicyEnforced, egressPolicyEnforced)
err := networkPolicy.Validate()
if err != nil {
return fmt.Errorf("error validating generated NetworkPolicy for %s: %s", ip, err), nil
Expand Down
64 changes: 54 additions & 10 deletions pkg/envoy/server_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (
"github.com/cilium/cilium/pkg/policy/api/kafka"
"github.com/cilium/cilium/pkg/proxy/logger"
"github.com/cilium/cilium/pkg/proxy/logger/test"
"github.com/cilium/cilium/pkg/u8proto"

cilium "github.com/cilium/proxy/go/cilium/api"
envoy_config_core "github.com/cilium/proxy/go/envoy/config/core/v3"
Expand Down Expand Up @@ -438,7 +439,7 @@ func (s *ServerSuite) TestGetDirectionNetworkPolicy(c *C) {
}

func (s *ServerSuite) TestGetNetworkPolicy(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4Policy1, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4Policy1, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand All @@ -450,7 +451,7 @@ func (s *ServerSuite) TestGetNetworkPolicy(c *C) {
}

func (s *ServerSuite) TestGetNetworkPolicyWildcard(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4Policy2, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4Policy2, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand All @@ -462,7 +463,7 @@ func (s *ServerSuite) TestGetNetworkPolicyWildcard(c *C) {
}

func (s *ServerSuite) TestGetNetworkPolicyDeny(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4Policy1RequiresV2, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4Policy1RequiresV2, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand All @@ -474,7 +475,7 @@ func (s *ServerSuite) TestGetNetworkPolicyDeny(c *C) {
}

func (s *ServerSuite) TestGetNetworkPolicyWildcardDeny(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4Policy1RequiresV2, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4Policy1RequiresV2, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand All @@ -486,7 +487,7 @@ func (s *ServerSuite) TestGetNetworkPolicyWildcardDeny(c *C) {
}

func (s *ServerSuite) TestGetNetworkPolicyNil(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, nil, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, nil, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand All @@ -498,7 +499,7 @@ func (s *ServerSuite) TestGetNetworkPolicyNil(c *C) {
}

func (s *ServerSuite) TestGetNetworkPolicyIngressNotEnforced(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4Policy2, false, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4Policy2, false, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand All @@ -510,7 +511,7 @@ func (s *ServerSuite) TestGetNetworkPolicyIngressNotEnforced(c *C) {
}

func (s *ServerSuite) TestGetNetworkPolicyEgressNotEnforced(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4Policy1RequiresV2, true, false)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4Policy1RequiresV2, true, false)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand Down Expand Up @@ -568,7 +569,7 @@ var ExpectedPerPortPoliciesL7 = []*cilium.PortNetworkPolicy{
}

func (s *ServerSuite) TestGetNetworkPolicyL7(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4PolicyL7, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4PolicyL7, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand Down Expand Up @@ -623,7 +624,7 @@ var ExpectedPerPortPoliciesKafka = []*cilium.PortNetworkPolicy{
}

func (s *ServerSuite) TestGetNetworkPolicyKafka(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4PolicyKafka, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4PolicyKafka, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand Down Expand Up @@ -694,7 +695,7 @@ var ExpectedPerPortPoliciesMySQL = []*cilium.PortNetworkPolicy{
}

func (s *ServerSuite) TestGetNetworkPolicyMySQL(c *C) {
obtained := getNetworkPolicy(ep, IPv4Addr, L4PolicyMySQL, true, true)
obtained := getNetworkPolicy(ep, nil, IPv4Addr, L4PolicyMySQL, true, true)
expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
Expand All @@ -703,3 +704,46 @@ func (s *ServerSuite) TestGetNetworkPolicyMySQL(c *C) {
}
c.Assert(obtained, checker.ExportedEquals, expected)
}

var L4PolicyL7Kafka = &policy.L4Policy{}

var nvp = &policy.VisibilityPolicy{
Ingress: make(policy.DirectionalVisibilityPolicy),
}

func (s *ServerSuite) TestGetNetworkPolicyProxylibVisibility(c *C) {

dvp := nvp.Ingress

pp := "9092/TCP"

dvp[pp] = &policy.VisibilityMetadata{ //"<Ingress/9092/TCP/Kafka>"
Port: 9092,
Parser: "Kafka",
Proto: u8proto.TCP,
Ingress: true,
L7Metadata: make(policy.L7DataMap),
}

pnp := []*cilium.PortNetworkPolicy{
{
Port: uint32(nvp.Ingress[pp].Port),
Protocol: envoy_config_core.SocketAddress_TCP,
Rules: []*cilium.PortNetworkPolicyRule{
{
L7Proto: nvp.Ingress[pp].Parser.String(),
},
},
},
}
obtained := getNetworkPolicy(ep, nvp, IPv4Addr, L4PolicyL7Kafka, false, false)

expected := &cilium.NetworkPolicy{
Name: IPv4Addr,
Policy: uint64(Identity),
IngressPerPortPolicies: pnp,
ConntrackMapName: "global",
}

c.Assert(obtained, checker.ExportedEquals, expected)
}
2 changes: 1 addition & 1 deletion pkg/fqdn/dnsproxy/proxy_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ func (s *DNSProxyTestSuite) RemoveProxyRedirect(e regeneration.EndpointInfoSourc
return nil, nil, nil
}

func (s *DNSProxyTestSuite) UpdateNetworkPolicy(e regeneration.EndpointUpdater, policy *policy.L4Policy,
func (s *DNSProxyTestSuite) UpdateNetworkPolicy(e regeneration.EndpointUpdater, vis *policy.VisibilityPolicy, policy *policy.L4Policy,
proxyWaitGroup *completion.WaitGroup) (error, revert.RevertFunc) {
return nil, nil
}
Expand Down
4 changes: 2 additions & 2 deletions pkg/proxy/proxy.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -636,8 +636,8 @@ func (p *Proxy) updateRedirectMetrics() {
}

// UpdateNetworkPolicy must update the redirect configuration of an endpoint in the proxy
func (p *Proxy) UpdateNetworkPolicy(ep logger.EndpointUpdater, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error) {
return p.XDSServer.UpdateNetworkPolicy(ep, policy, ingressPolicyEnforced, egressPolicyEnforced, wg)
func (p *Proxy) UpdateNetworkPolicy(ep logger.EndpointUpdater, vis *policy.VisibilityPolicy, policy *policy.L4Policy, ingressPolicyEnforced, egressPolicyEnforced bool, wg *completion.WaitGroup) (error, func() error) {
return p.XDSServer.UpdateNetworkPolicy(ep, vis, policy, ingressPolicyEnforced, egressPolicyEnforced, wg)
}

// UseCurrentNetworkPolicy inserts a Completion to the WaitGroup if the current network policy has not yet been acked
Expand Down

0 comments on commit 54b81ae

Please sign in to comment.