Intentions ACL enforcement updates (#7028)

* Renamed structs.IntentionWildcard to structs.WildcardSpecifier * Refactor ACL Config Get rid of remnants of enterprise only renaming. Add a WildcardName field for specifying what string should be used to indicate a wildcard. * Add wildcard support in the ACL package For read operations they can call anyAllowed to determine if any read access to the given resource would be granted. For write operations they can call allAllowed to ensure that write access is granted to everything. * Make v1/agent/connect/authorize namespace aware * Update intention ACL enforcement This also changes how intention:read is granted. Before the Intention.List RPC would allow viewing an intention if the token had intention:read on the destination. However Intention.Match allowed viewing if access was allowed for either the source or dest side. Now Intention.List and Intention.Get fall in line with Intention.Matches previous behavior. Due to this being done a few different places ACL enforcement for a singular intention is now done with the CanRead and CanWrite methods on the intention itself. * Refactor Intention.Apply to make things easier to follow.
2020-01-13 15:51:40 -05:00 · 2020-01-13 15:51:40 -05:00 · baa89c7c65
parent 61fc4f8253
commit baa89c7c65
27 changed files with 1570 additions and 261 deletions
--- a/acl/acl.go
+++ b/acl/acl.go
@ -0,0 +1,32 @@
+package acl
+
+const (
+	WildcardName = "*"
+)
+
+// Config encapsualtes all of the generic configuration parameters used for
+// policy parsing and enforcement
+type Config struct {
+	// WildcardName is the string that represents a request to authorize a wildcard permission
+	WildcardName string
+
+	// embedded enterprise configuration
+	EnterpriseConfig
+}
+
+// GetWildcardName will retrieve the configured wildcard name or provide a default
+// in the case that the config is Nil or the wildcard name is unset.
+func (c *Config) GetWildcardName() string {
+	if c == nil || c.WildcardName == "" {
+		return WildcardName
+	}
+	return c.WildcardName
+}
+
+// Close will relinquish any resources this Config might be holding on to or
+// managing.
+func (c *Config) Close() {
+	if c != nil {
+		c.EnterpriseConfig.Close()
+	}
+}
--- a/acl/acl_oss.go
+++ b/acl/acl_oss.go
@ -2,7 +2,10 @@

 package acl

-// Config stub
-type Config struct{}
+type EnterpriseConfig struct {
+	// no fields in OSS
+}

-func (_ *Config) Close() {}
+func (_ *EnterpriseConfig) Close() {
+	// do nothing
+}
--- a/acl/authorizer.go
+++ b/acl/authorizer.go
@ -242,3 +242,14 @@ func Enforce(authz Authorizer, rsc Resource, segment string, access string, ctx

 	return Deny, fmt.Errorf("Invalid access level for %s resource: %s", rsc, access)
 }
+
+// NewAuthorizerFromRules is a convenience function to invoke NewPolicyFromSource followed by NewPolicyAuthorizer with
+// the parse policy.
+func NewAuthorizerFromRules(id string, revision uint64, rules string, syntax SyntaxVersion, conf *Config, meta *EnterprisePolicyMeta) (Authorizer, error) {
+	policy, err := NewPolicyFromSource(id, revision, rules, syntax, conf, meta)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewPolicyAuthorizer([]*Policy{policy}, conf)
+}
--- a/acl/policy_authorizer.go
+++ b/acl/policy_authorizer.go
@ -340,6 +340,107 @@ func newPolicyAuthorizerFromRules(rules *PolicyRules, ent *Config) (Authorizer,
 	return p, nil
 }

+// enforceCallbacks are to be passed to anyAllowed or allAllowed. The interface{}
+// parameter will be a value stored in the radix.Tree passed to those functions.
+// prefixOnly indicates that only we only want to consider the prefix matching rule
+// if any. The return value indicates whether this one leaf node in the tree would
+// allow, deny or make no decision regarding some authorization.
+type enforceCallback func(raw interface{}, prefixOnly bool) EnforcementDecision
+
+func anyAllowed(tree *radix.Tree, enforceFn enforceCallback) EnforcementDecision {
+	decision := Default
+
+	// special case for handling a catch-all prefix rule. If the rule woul Deny access then our default decision
+	// should be to Deny, but this decision should still be overridable with other more specific rules.
+	if raw, found := tree.Get(""); found {
+		decision = enforceFn(raw, true)
+		if decision == Allow {
+			return Allow
+		}
+	}
+
+	tree.Walk(func(path string, raw interface{}) bool {
+		if enforceFn(raw, false) == Allow {
+			decision = Allow
+			return true
+		}
+
+		return false
+	})
+
+	return decision
+}
+
+func allAllowed(tree *radix.Tree, enforceFn enforceCallback) EnforcementDecision {
+	decision := Default
+
+	// look for a "" prefix rule
+	if raw, found := tree.Get(""); found {
+		// ensure that the empty prefix rule would allow the access
+		// if it does allow it we still must check all the other rules to ensure
+		// nothing overrides the top level grant with a different access level
+		// if not we can return early
+		decision = enforceFn(raw, true)
+
+		// the top level prefix rule denied access so we can return early.
+		if decision == Deny {
+			return Deny
+		}
+	}
+
+	tree.Walk(func(path string, raw interface{}) bool {
+		if enforceFn(raw, false) == Deny {
+			decision = Deny
+			return true
+		}
+		return false
+	})
+
+	return decision
+}
+
+func (authz *policyAuthorizer) anyAllowed(tree *radix.Tree, requiredPermission AccessLevel) EnforcementDecision {
+	return anyAllowed(tree, func(raw interface{}, prefixOnly bool) EnforcementDecision {
+		leaf := raw.(*policyAuthorizerRadixLeaf)
+		decision := Default
+
+		if leaf.prefix != nil {
+			decision = enforce(leaf.prefix.access, requiredPermission)
+		}
+
+		if prefixOnly || decision == Allow || leaf.exact == nil {
+			return decision
+		}
+
+		return enforce(leaf.exact.access, requiredPermission)
+	})
+}
+
+func (authz *policyAuthorizer) allAllowed(tree *radix.Tree, requiredPermission AccessLevel) EnforcementDecision {
+	return allAllowed(tree, func(raw interface{}, prefixOnly bool) EnforcementDecision {
+		leaf := raw.(*policyAuthorizerRadixLeaf)
+		prefixDecision := Default
+
+		if leaf.prefix != nil {
+			prefixDecision = enforce(leaf.prefix.access, requiredPermission)
+		}
+
+		if prefixOnly || prefixDecision == Deny || leaf.exact == nil {
+			return prefixDecision
+		}
+
+		decision := enforce(leaf.exact.access, requiredPermission)
+
+		if decision == Default {
+			// basically this means defer to the prefix decision as the
+			// authorizer rule made no decision with an exact match rule
+			return prefixDecision
+		}
+
+		return decision
+	})
+}
+
 // ACLRead checks if listing of ACLs is allowed
 func (p *policyAuthorizer) ACLRead(*AuthorizerContext) EnforcementDecision {
 	if p.aclRule != nil {
@ -410,6 +511,10 @@ func (p *policyAuthorizer) IntentionDefaultAllow(_ *AuthorizerContext) Enforceme
 // IntentionRead checks if writing (creating, updating, or deleting) of an
 // intention is allowed.
 func (p *policyAuthorizer) IntentionRead(prefix string, _ *AuthorizerContext) EnforcementDecision {
+	if prefix == "*" {
+		return p.anyAllowed(p.intentionRules, AccessRead)
+	}
+
 	if rule, ok := getPolicy(prefix, p.intentionRules); ok {
 		return enforce(rule.access, AccessRead)
 	}
@ -419,6 +524,10 @@ func (p *policyAuthorizer) IntentionRead(prefix string, _ *AuthorizerContext) En
 // IntentionWrite checks if writing (creating, updating, or deleting) of an
 // intention is allowed.
 func (p *policyAuthorizer) IntentionWrite(prefix string, _ *AuthorizerContext) EnforcementDecision {
+	if prefix == "*" {
+		return p.allAllowed(p.intentionRules, AccessWrite)
+	}
+
 	if rule, ok := getPolicy(prefix, p.intentionRules); ok {
 		return enforce(rule.access, AccessWrite)
 	}
--- a/acl/policy_authorizer_test.go
+++ b/acl/policy_authorizer_test.go
@ -4,6 +4,7 @@ import (
 	"fmt"
 	"testing"

+	"github.com/armon/go-radix"
 	"github.com/stretchr/testify/require"
 )

@ -343,6 +344,102 @@ func TestPolicyAuthorizer(t *testing.T) {
 				{name: "PreparedQueryWriteDenied", prefix: "football", check: checkDenyPreparedQueryWrite},
 			},
 		},
+		"Intention Wildcards - prefix denied": aclTest{
+			policy: &Policy{PolicyRules: PolicyRules{
+				Services: []*ServiceRule{
+					&ServiceRule{
+						Name:       "foo",
+						Policy:     PolicyWrite,
+						Intentions: PolicyWrite,
+					},
+				},
+				ServicePrefixes: []*ServiceRule{
+					&ServiceRule{
+						Name:       "",
+						Policy:     PolicyDeny,
+						Intentions: PolicyDeny,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "AnyAllowed", prefix: "*", check: checkAllowIntentionRead},
+				{name: "AllDenied", prefix: "*", check: checkDenyIntentionWrite},
+			},
+		},
+		"Intention Wildcards - prefix allowed": aclTest{
+			policy: &Policy{PolicyRules: PolicyRules{
+				Services: []*ServiceRule{
+					&ServiceRule{
+						Name:       "foo",
+						Policy:     PolicyWrite,
+						Intentions: PolicyDeny,
+					},
+				},
+				ServicePrefixes: []*ServiceRule{
+					&ServiceRule{
+						Name:       "",
+						Policy:     PolicyWrite,
+						Intentions: PolicyWrite,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "AnyAllowed", prefix: "*", check: checkAllowIntentionRead},
+				{name: "AllDenied", prefix: "*", check: checkDenyIntentionWrite},
+			},
+		},
+		"Intention Wildcards - all allowed": aclTest{
+			policy: &Policy{PolicyRules: PolicyRules{
+				Services: []*ServiceRule{
+					&ServiceRule{
+						Name:       "foo",
+						Policy:     PolicyWrite,
+						Intentions: PolicyWrite,
+					},
+				},
+				ServicePrefixes: []*ServiceRule{
+					&ServiceRule{
+						Name:       "",
+						Policy:     PolicyWrite,
+						Intentions: PolicyWrite,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "AnyAllowed", prefix: "*", check: checkAllowIntentionRead},
+				{name: "AllAllowed", prefix: "*", check: checkAllowIntentionWrite},
+			},
+		},
+		"Intention Wildcards - all default": aclTest{
+			policy: &Policy{PolicyRules: PolicyRules{
+				Services: []*ServiceRule{
+					&ServiceRule{
+						Name:       "foo",
+						Policy:     PolicyWrite,
+						Intentions: PolicyWrite,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "AnyAllowed", prefix: "*", check: checkAllowIntentionRead},
+				{name: "AllDefault", prefix: "*", check: checkDefaultIntentionWrite},
+			},
+		},
+		"Intention Wildcards - any default": aclTest{
+			policy: &Policy{PolicyRules: PolicyRules{
+				Services: []*ServiceRule{
+					&ServiceRule{
+						Name:       "foo",
+						Policy:     PolicyWrite,
+						Intentions: PolicyDeny,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "AnyDefault", prefix: "*", check: checkDefaultIntentionRead},
+				{name: "AllDenied", prefix: "*", check: checkDenyIntentionWrite},
+			},
+		},
 	}

 	for name, tcase := range cases {
@ -369,3 +466,498 @@ func TestPolicyAuthorizer(t *testing.T) {
 		})
 	}
 }
+
+func TestAnyAllowed(t *testing.T) {
+	t.Parallel()
+
+	type radixInsertion struct {
+		segment string
+		value   *policyAuthorizerRadixLeaf
+	}
+
+	type testCase struct {
+		insertions []radixInsertion
+
+		readEnforcement  EnforcementDecision
+		listEnforcement  EnforcementDecision
+		writeEnforcement EnforcementDecision
+	}
+
+	cases := map[string]testCase{
+		"no-rules-default": testCase{
+			readEnforcement:  Default,
+			listEnforcement:  Default,
+			writeEnforcement: Default,
+		},
+		"prefix-write-allowed": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				// this shouldn't affect whether anyAllowed returns things are allowed
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Allow,
+		},
+		"prefix-list-allowed": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessList},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Deny,
+		},
+		"prefix-read-allowed": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessRead},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+			},
+			readEnforcement:  Deny,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny-other-write-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Allow,
+		},
+		"prefix-deny-other-write-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Allow,
+		},
+		"prefix-deny-other-list-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessList},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny-other-list-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessList},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny-other-read-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessRead},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny-other-read-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessRead},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny-other-deny-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+			},
+			readEnforcement:  Deny,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny-other-deny-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+			},
+			readEnforcement:  Deny,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+	}
+
+	for name, tcase := range cases {
+		t.Run(name, func(t *testing.T) {
+			tree := radix.New()
+
+			for _, insertion := range tcase.insertions {
+				tree.Insert(insertion.segment, insertion.value)
+			}
+
+			var authz policyAuthorizer
+			require.Equal(t, tcase.readEnforcement, authz.anyAllowed(tree, AccessRead))
+			require.Equal(t, tcase.listEnforcement, authz.anyAllowed(tree, AccessList))
+			require.Equal(t, tcase.writeEnforcement, authz.anyAllowed(tree, AccessWrite))
+		})
+	}
+}
+
+func TestAllAllowed(t *testing.T) {
+	t.Parallel()
+
+	type radixInsertion struct {
+		segment string
+		value   *policyAuthorizerRadixLeaf
+	}
+
+	type testCase struct {
+		insertions []radixInsertion
+
+		readEnforcement  EnforcementDecision
+		listEnforcement  EnforcementDecision
+		writeEnforcement EnforcementDecision
+	}
+
+	cases := map[string]testCase{
+		"no-rules-default": testCase{
+			readEnforcement:  Default,
+			listEnforcement:  Default,
+			writeEnforcement: Default,
+		},
+		"prefix-write-allowed": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Allow,
+		},
+		"prefix-list-allowed": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessList},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Deny,
+		},
+		"prefix-read-allowed": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessRead},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-deny": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+			},
+			readEnforcement:  Deny,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-allow-other-write-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Allow,
+		},
+		"prefix-allow-other-write-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Allow,
+		},
+		"prefix-allow-other-list-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessList},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Deny,
+		},
+		"prefix-allow-other-list-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessList},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Allow,
+			writeEnforcement: Deny,
+		},
+		"prefix-allow-other-read-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessRead},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-allow-other-read-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessRead},
+					},
+				},
+			},
+			readEnforcement:  Allow,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-allow-other-deny-prefix": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+			},
+			readEnforcement:  Deny,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+		"prefix-allow-other-deny-exact": testCase{
+			insertions: []radixInsertion{
+				radixInsertion{
+					segment: "",
+					value: &policyAuthorizerRadixLeaf{
+						prefix: &policyAuthorizerRule{access: AccessWrite},
+					},
+				},
+				radixInsertion{
+					segment: "foo",
+					value: &policyAuthorizerRadixLeaf{
+						exact: &policyAuthorizerRule{access: AccessDeny},
+					},
+				},
+			},
+			readEnforcement:  Deny,
+			listEnforcement:  Deny,
+			writeEnforcement: Deny,
+		},
+	}
+
+	for name, tcase := range cases {
+		t.Run(name, func(t *testing.T) {
+			tree := radix.New()
+
+			for _, insertion := range tcase.insertions {
+				tree.Insert(insertion.segment, insertion.value)
+			}
+
+			var authz policyAuthorizer
+			require.Equal(t, tcase.readEnforcement, authz.allAllowed(tree, AccessRead))
+			require.Equal(t, tcase.listEnforcement, authz.allAllowed(tree, AccessList))
+			require.Equal(t, tcase.writeEnforcement, authz.allAllowed(tree, AccessWrite))
+		})
+	}
+}
--- a/agent/agent_endpoint.go
+++ b/agent/agent_endpoint.go
@ -1360,9 +1360,12 @@ func (s *HTTPServer) AgentConnectAuthorize(resp http.ResponseWriter, req *http.R
 	var token string
 	s.parseToken(req, &token)

-	// TODO (namespaces) probably need an update here to include the namespace with the target in the request
-	// Decode the request from the request body
 	var authReq structs.ConnectAuthorizeRequest
+
+	if err := s.parseEntMetaNoWildcard(req, &authReq.EnterpriseMeta); err != nil {
+		return nil, err
+	}
+
 	if err := decodeBody(req.Body, &authReq); err != nil {
 		return nil, BadRequestError{fmt.Sprintf("Request decode failed: %v", err)}
 	}
--- a/agent/connect/uri_service.go
+++ b/agent/connect/uri_service.go
@ -27,12 +27,12 @@ func (id *SpiffeIDService) URI() *url.URL {

 // CertURI impl.
 func (id *SpiffeIDService) Authorize(ixn *structs.Intention) (bool, bool) {
-	if ixn.SourceNS != structs.IntentionWildcard && ixn.SourceNS != id.Namespace {
+	if ixn.SourceNS != structs.WildcardSpecifier && ixn.SourceNS != id.Namespace {
 		// Non-matching namespace
 		return false, false
 	}

-	if ixn.SourceName != structs.IntentionWildcard && ixn.SourceName != id.Service {
+	if ixn.SourceName != structs.WildcardSpecifier && ixn.SourceName != id.Service {
 		// Non-matching name
 		return false, false
 	}
--- a/agent/connect/uri_service_test.go
+++ b/agent/connect/uri_service_test.go
@ -74,7 +74,7 @@ func TestSpiffeIDServiceAuthorize(t *testing.T) {
 			serviceWeb,
 			&structs.Intention{
 				SourceNS:   serviceWeb.Namespace,
-				SourceName: structs.IntentionWildcard,
+				SourceName: structs.WildcardSpecifier,
 				Action:     structs.IntentionActionDeny,
 			},
 			false,
@ -86,7 +86,7 @@ func TestSpiffeIDServiceAuthorize(t *testing.T) {
 			serviceWeb,
 			&structs.Intention{
 				SourceNS:   serviceWeb.Namespace,
-				SourceName: structs.IntentionWildcard,
+				SourceName: structs.WildcardSpecifier,
 				Action:     structs.IntentionActionAllow,
 			},
 			true,
--- a/agent/connect_auth.go
+++ b/agent/connect_auth.go
@ -22,7 +22,7 @@ import (
 // error is returned, otherwise error indicates an unexpected server failure. If
 // access is denied, no error is returned but the first return value is false.
 func (a *Agent) ConnectAuthorize(token string,
-	req *structs.ConnectAuthorizeRequest) (authz bool, reason string, m *cache.ResultMeta, err error) {
+	req *structs.ConnectAuthorizeRequest) (allowed bool, reason string, m *cache.ResultMeta, err error) {

 	// Helper to make the error cases read better without resorting to named
 	// returns which get messy and prone to mistakes in a method this long.
@ -53,12 +53,13 @@ func (a *Agent) ConnectAuthorize(token string,
 	// We need to verify service:write permissions for the given token.
 	// We do this manually here since the RPC request below only verifies
 	// service:read.
-	rule, err := a.resolveToken(token)
+	var authzContext acl.AuthorizerContext
+	authz, err := a.resolveTokenAndDefaultMeta(token, &req.EnterpriseMeta, &authzContext)
 	if err != nil {
 		return returnErr(err)
 	}
-	// TODO (namespaces) - pass through a real ent authz ctx
-	if rule != nil && rule.ServiceWrite(req.Target, nil) != acl.Allow {
+
+	if authz != nil && authz.ServiceWrite(req.Target, &authzContext) != acl.Allow {
 		return returnErr(acl.ErrPermissionDenied)
 	}

@ -74,7 +75,7 @@ func (a *Agent) ConnectAuthorize(token string,
 			Type: structs.IntentionMatchDestination,
 			Entries: []structs.IntentionMatchEntry{
 				{
-					Namespace: structs.IntentionDefaultNamespace,
+					Namespace: req.TargetNamespace(),
 					Name:      req.Target,
 				},
 			},
@ -107,15 +108,14 @@ func (a *Agent) ConnectAuthorize(token string,
 	// specifying the anonymous token, which will get the default behavior. The
 	// default behavior if ACLs are disabled is to allow connections to mimic the
 	// behavior of Consul itself: everything is allowed if ACLs are disabled.
-	rule, err = a.resolveToken("")
+	authz, err = a.resolveToken("")
 	if err != nil {
 		return returnErr(err)
 	}
-	if rule == nil {
+	if authz == nil {
 		// ACLs not enabled at all, the default is allow all.
 		return true, "ACLs disabled, access is allowed by default", &meta, nil
 	}
 	reason = "Default behavior configured by ACLs"
-	// TODO (namespaces) - pass through a real ent authz ctx
-	return rule.IntentionDefaultAllow(nil) == acl.Allow, reason, &meta, nil
+	return authz.IntentionDefaultAllow(nil) == acl.Allow, reason, &meta, nil
 }
--- a/agent/consul/acl.go
+++ b/agent/consul/acl.go
@ -167,8 +167,9 @@ type ACLResolverConfig struct {
 	// so that it can detect when the servers have gotten ACLs enabled.
 	AutoDisable bool

-	// EnterpriseACLConfig contains Consul Enterprise specific ACL configuration
-	EnterpriseConfig *acl.Config
+	// ACLConfig is the configuration necessary to pass through to the acl package when creating authorizers
+	// and when authorizing access
+	ACLConfig *acl.Config
 }

 // ACLResolver is the type to handle all your token and policy resolution needs.
@ -201,7 +202,7 @@ type ACLResolver struct {
 	logger *log.Logger

 	delegate ACLResolverDelegate
-	entConf  *acl.Config
+	aclConf  *acl.Config

 	cache         *structs.ACLCaches
 	identityGroup singleflight.Group
@ -254,7 +255,7 @@ func NewACLResolver(config *ACLResolverConfig) (*ACLResolver, error) {
 		config:      config.Config,
 		logger:      config.Logger,
 		delegate:    config.Delegate,
-		entConf:     config.EnterpriseConfig,
+		aclConf:     config.ACLConfig,
 		cache:       cache,
 		autoDisable: config.AutoDisable,
 		down:        down,
@ -262,7 +263,7 @@ func NewACLResolver(config *ACLResolverConfig) (*ACLResolver, error) {
 }

 func (r *ACLResolver) Close() {
-	r.entConf.Close()
+	r.aclConf.Close()
 }

 func (r *ACLResolver) fetchAndCacheTokenLegacy(token string, cached *structs.AuthorizerCacheEntry) (acl.Authorizer, error) {
@ -295,7 +296,7 @@ func (r *ACLResolver) fetchAndCacheTokenLegacy(token string, cached *structs.Aut
 			policies = append(policies, policy.ConvertFromLegacy())
 		}

-		authorizer, err := acl.NewPolicyAuthorizerWithDefaults(parent, policies, r.entConf)
+		authorizer, err := acl.NewPolicyAuthorizerWithDefaults(parent, policies, r.aclConf)

 		r.cache.PutAuthorizerWithTTL(token, authorizer, reply.TTL)
 		return authorizer, err
@ -338,7 +339,7 @@ func (r *ACLResolver) resolveTokenLegacy(token string) (structs.ACLIdentity, acl
 				return identity, nil, err
 			}

-			authz, err := policies.Compile(r.cache, r.entConf)
+			authz, err := policies.Compile(r.cache, r.aclConf)
 			if err != nil {
 				return identity, nil, err
 			}
@ -1065,7 +1066,7 @@ func (r *ACLResolver) ResolveTokenToIdentityAndAuthorizer(token string) (structs
 	// Build the Authorizer
 	var chain []acl.Authorizer

-	authz, err := policies.Compile(r.cache, r.entConf)
+	authz, err := policies.Compile(r.cache, r.aclConf)
 	if err != nil {
 		return nil, nil, err
 	}
@ -1116,7 +1117,7 @@ func (r *ACLResolver) GetMergedPolicyForToken(token string) (*acl.Policy, error)
 		return nil, acl.ErrNotFound
 	}

-	return policies.Merge(r.cache, r.entConf)
+	return policies.Merge(r.cache, r.aclConf)
 }

 // aclFilter is used to filter results from our state store based on ACL rules
@ -1343,21 +1344,9 @@ func (f *aclFilter) filterCoordinates(coords *structs.Coordinates) {
 // We prune entries the user doesn't have access to, and we redact any tokens
 // if the user doesn't have a management token.
 func (f *aclFilter) filterIntentions(ixns *structs.Intentions) {
-	// Otherwise, we need to see what the token has access to.
 	ret := make(structs.Intentions, 0, len(*ixns))
 	for _, ixn := range *ixns {
-		// TODO (namespaces) update to call with an actual ent authz context once connect supports it
-		// This probably should get translated into multiple calls where having acl:read in either the
-		// source or destination namespace is enough to grant read on the intention
-		aclRead := f.authorizer.ACLRead(nil) == acl.Allow
-
-		// If no prefix ACL applies to this then filter it, since
-		// we know at this point the user doesn't have a management
-		// token, otherwise see what the policy says.
-		prefix, ok := ixn.GetACLPrefix()
-
-		// TODO (namespaces) update to call with an actual ent authz context once connect supports it
-		if !aclRead && (!ok || f.authorizer.IntentionRead(prefix, nil) != acl.Allow) {
+		if !ixn.CanRead(f.authorizer) {
 			f.logger.Printf("[DEBUG] consul: dropping intention %q from result due to ACLs", ixn.ID)
 			continue
 		}
--- a/agent/consul/acl_endpoint.go
+++ b/agent/consul/acl_endpoint.go
@ -1070,7 +1070,7 @@ func (a *ACL) PolicySet(args *structs.ACLPolicySetRequest, reply *structs.ACLPol
 	}

 	// validate the rules
-	_, err = acl.NewPolicyFromSource("", 0, policy.Rules, policy.Syntax, a.srv.enterpriseACLConfig, policy.EnterprisePolicyMeta())
+	_, err = acl.NewPolicyFromSource("", 0, policy.Rules, policy.Syntax, a.srv.aclConfig, policy.EnterprisePolicyMeta())
 	if err != nil {
 		return err
 	}
--- a/agent/consul/acl_endpoint_legacy.go
+++ b/agent/consul/acl_endpoint_legacy.go
@ -114,7 +114,7 @@ func aclApplyInternal(srv *Server, args *structs.ACLRequest, reply *string) erro
 		}

 		// Validate the rules compile
-		_, err := acl.NewPolicyFromSource("", 0, args.ACL.Rules, acl.SyntaxLegacy, srv.enterpriseACLConfig, nil)
+		_, err := acl.NewPolicyFromSource("", 0, args.ACL.Rules, acl.SyntaxLegacy, srv.aclConfig, nil)
 		if err != nil {
 			return fmt.Errorf("ACL rule compilation failed: %v", err)
 		}
--- a/agent/consul/acl_oss.go
+++ b/agent/consul/acl_oss.go
@ -16,8 +16,10 @@ func (s *Server) replicationEnterpriseMeta() *structs.EnterpriseMeta {
 	return structs.ReplicationEnterpriseMeta()
 }

-func newEnterpriseACLConfig(*log.Logger) *acl.Config {
-	return nil
+func newACLConfig(*log.Logger) *acl.Config {
+	return &acl.Config{
+		WildcardName: structs.WildcardSpecifier,
+	}
 }

 func (r *ACLResolver) resolveEnterpriseDefaultsForIdentity(identity structs.ACLIdentity) (acl.Authorizer, error) {
--- a/agent/consul/client.go
+++ b/agent/consul/client.go
@ -154,12 +154,12 @@ func NewClientLogger(config *Config, logger *log.Logger, tlsConfigurator *tlsuti

 	c.useNewACLs = 0
 	aclConfig := ACLResolverConfig{
-		Config:           config,
-		Delegate:         c,
-		Logger:           logger,
-		AutoDisable:      true,
-		CacheConfig:      clientACLCacheConfig,
-		EnterpriseConfig: newEnterpriseACLConfig(logger),
+		Config:      config,
+		Delegate:    c,
+		Logger:      logger,
+		AutoDisable: true,
+		CacheConfig: clientACLCacheConfig,
+		ACLConfig:   newACLConfig(logger),
 	}
 	var err error
 	if c.acls, err = NewACLResolver(&aclConfig); err != nil {
--- a/agent/consul/intention_endpoint.go
+++ b/agent/consul/intention_endpoint.go
@ -10,8 +10,8 @@ import (
 	"github.com/hashicorp/consul/agent/connect"
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/go-memdb"
-	"github.com/hashicorp/go-uuid"
 )

 var (
@ -25,6 +25,137 @@ type Intention struct {
 	srv *Server
 }

+func (s *Intention) checkIntentionID(id string) (bool, error) {
+	state := s.srv.fsm.State()
+	if _, ixn, err := state.IntentionGet(nil, id); err != nil {
+		return false, err
+	} else if ixn != nil {
+		return false, nil
+	}
+
+	return true, nil
+}
+
+// prepareApplyCreate validates that the requester has permissions to create the new intention,
+// generates a new uuid for the intention and generally validates that the request is well-formed
+func (s *Intention) prepareApplyCreate(authz acl.Authorizer, entMeta *structs.EnterpriseMeta, args *structs.IntentionRequest) error {
+	if !args.Intention.CanWrite(authz) {
+		s.srv.logger.Printf("[WARN] consul.intention: Intention creation denied due to ACLs")
+		return acl.ErrPermissionDenied
+	}
+
+	// If no ID is provided, generate a new ID. This must be done prior to
+	// appending to the Raft log, because the ID is not deterministic. Once
+	// the entry is in the log, the state update MUST be deterministic or
+	// the followers will not converge.
+	if args.Intention.ID != "" {
+		return fmt.Errorf("ID must be empty when creating a new intention")
+	}
+
+	var err error
+	args.Intention.ID, err = lib.GenerateUUID(s.checkIntentionID)
+	if err != nil {
+		return err
+	}
+	// Set the created at
+	args.Intention.CreatedAt = time.Now().UTC()
+	args.Intention.UpdatedAt = args.Intention.CreatedAt
+
+	// Default source type
+	if args.Intention.SourceType == "" {
+		args.Intention.SourceType = structs.IntentionSourceConsul
+	}
+
+	args.Intention.DefaultNamespaces(entMeta)
+
+	// Validate. We do not validate on delete since it is valid to only
+	// send an ID in that case.
+	// Set the precedence
+	args.Intention.UpdatePrecedence()
+
+	if err := args.Intention.Validate(); err != nil {
+		return err
+	}
+
+	// make sure we set the hash prior to raft application
+	args.Intention.SetHash(true)
+
+	return nil
+}
+
+// prepareApplyUpdate validates that the requester has permissions on both the updated and existing
+// intention as well as generally validating that the request is well-formed
+func (s *Intention) prepareApplyUpdate(authz acl.Authorizer, entMeta *structs.EnterpriseMeta, args *structs.IntentionRequest) error {
+	if !args.Intention.CanWrite(authz) {
+		s.srv.logger.Printf("[WARN] consul.intention: Update operation on intention %q denied due to ACLs", args.Intention.ID)
+		return acl.ErrPermissionDenied
+	}
+
+	_, ixn, err := s.srv.fsm.State().IntentionGet(nil, args.Intention.ID)
+	if err != nil {
+		return fmt.Errorf("Intention lookup failed: %v", err)
+	}
+	if ixn == nil {
+		return fmt.Errorf("Cannot modify non-existent intention: '%s'", args.Intention.ID)
+	}
+
+	// Perform the ACL check that we have write to the old intention too,
+	// which must be true to perform any rename. This is the only ACL enforcement
+	// done for deletions and a secondary enforcement for updates.
+	if !ixn.CanWrite(authz) {
+		s.srv.logger.Printf("[WARN] consul.intention: Update operation on intention %q denied due to ACLs", args.Intention.ID)
+		return acl.ErrPermissionDenied
+	}
+
+	// We always update the updatedat field.
+	args.Intention.UpdatedAt = time.Now().UTC()
+
+	// Default source type
+	if args.Intention.SourceType == "" {
+		args.Intention.SourceType = structs.IntentionSourceConsul
+	}
+
+	args.Intention.DefaultNamespaces(entMeta)
+
+	// Validate. We do not validate on delete since it is valid to only
+	// send an ID in that case.
+	// Set the precedence
+	args.Intention.UpdatePrecedence()
+
+	if err := args.Intention.Validate(); err != nil {
+		return err
+	}
+
+	// make sure we set the hash prior to raft application
+	args.Intention.SetHash(true)
+
+	return nil
+}
+
+// prepareApplyDelete ensures that the intention specified by the ID in the request exists
+// and that the requester is authorized to delete it
+func (s *Intention) prepareApplyDelete(authz acl.Authorizer, entMeta *structs.EnterpriseMeta, args *structs.IntentionRequest) error {
+	// If this is not a create, then we have to verify the ID.
+	state := s.srv.fsm.State()
+	_, ixn, err := state.IntentionGet(nil, args.Intention.ID)
+	if err != nil {
+		return fmt.Errorf("Intention lookup failed: %v", err)
+	}
+	if ixn == nil {
+		return fmt.Errorf("Cannot delete non-existent intention: '%s'", args.Intention.ID)
+	}
+
+	// Perform the ACL check that we have write to the old intention too,
+	// which must be true to perform any rename. This is the only ACL enforcement
+	// done for deletions and a secondary enforcement for updates.
+	if !ixn.CanWrite(authz) {
+		s.srv.logger.Printf("[WARN] consul.intention: Deletion operation on intention %q denied due to ACLs", args.Intention.ID)
+		return acl.ErrPermissionDenied
+	}
+
+	return nil
+}
+
 // Apply creates or updates an intention in the data store.
 func (s *Intention) Apply(
 	args *structs.IntentionRequest,
@ -46,103 +177,32 @@ func (s *Intention) Apply(
 		args.Intention = &structs.Intention{}
 	}

-	// If no ID is provided, generate a new ID. This must be done prior to
-	// appending to the Raft log, because the ID is not deterministic. Once
-	// the entry is in the log, the state update MUST be deterministic or
-	// the followers will not converge.
-	if args.Op == structs.IntentionOpCreate {
-		if args.Intention.ID != "" {
-			return fmt.Errorf("ID must be empty when creating a new intention")
-		}
-
-		state := s.srv.fsm.State()
-		for {
-			var err error
-			args.Intention.ID, err = uuid.GenerateUUID()
-			if err != nil {
-				s.srv.logger.Printf("[ERR] consul.intention: UUID generation failed: %v", err)
-				return err
-			}
-
-			_, ixn, err := state.IntentionGet(nil, args.Intention.ID)
-			if err != nil {
-				s.srv.logger.Printf("[ERR] consul.intention: intention lookup failed: %v", err)
-				return err
-			}
-			if ixn == nil {
-				break
-			}
-		}
-
-		// Set the created at
-		args.Intention.CreatedAt = time.Now().UTC()
-	}
-	*reply = args.Intention.ID
-
 	// Get the ACL token for the request for the checks below.
-	rule, err := s.srv.ResolveToken(args.Token)
+	var entMeta structs.EnterpriseMeta
+	authz, err := s.srv.ResolveTokenAndDefaultMeta(args.Token, &entMeta, nil)
 	if err != nil {
 		return err
 	}

-	// Perform the ACL check
-	if prefix, ok := args.Intention.GetACLPrefix(); ok {
-		if rule != nil && rule.IntentionWrite(prefix, nil) != acl.Allow {
-			s.srv.logger.Printf("[WARN] consul.intention: Operation on intention '%s' denied due to ACLs", args.Intention.ID)
-			return acl.ErrPermissionDenied
-		}
-	}
-
-	// If this is not a create, then we have to verify the ID.
-	if args.Op != structs.IntentionOpCreate {
-		state := s.srv.fsm.State()
-		_, ixn, err := state.IntentionGet(nil, args.Intention.ID)
-		if err != nil {
-			return fmt.Errorf("Intention lookup failed: %v", err)
-		}
-		if ixn == nil {
-			return fmt.Errorf("Cannot modify non-existent intention: '%s'", args.Intention.ID)
-		}
-
-		// Perform the ACL check that we have write to the old prefix too,
-		// which must be true to perform any rename.
-		if prefix, ok := ixn.GetACLPrefix(); ok {
-			if rule != nil && rule.IntentionWrite(prefix, nil) != acl.Allow {
-				s.srv.logger.Printf("[WARN] consul.intention: Operation on intention '%s' denied due to ACLs", args.Intention.ID)
-				return acl.ErrPermissionDenied
-			}
-		}
-	}
-
-	// We always update the updatedat field. This has no effect for deletion.
-	args.Intention.UpdatedAt = time.Now().UTC()
-
-	// Default source type
-	if args.Intention.SourceType == "" {
-		args.Intention.SourceType = structs.IntentionSourceConsul
-	}
-
-	// Until we support namespaces, we force all namespaces to be default
-	if args.Intention.SourceNS == "" {
-		args.Intention.SourceNS = structs.IntentionDefaultNamespace
-	}
-	if args.Intention.DestinationNS == "" {
-		args.Intention.DestinationNS = structs.IntentionDefaultNamespace
-	}
-
-	// Validate. We do not validate on delete since it is valid to only
-	// send an ID in that case.
-	if args.Op != structs.IntentionOpDelete {
-		// Set the precedence
-		args.Intention.UpdatePrecedence()
-
-		if err := args.Intention.Validate(); err != nil {
+	switch args.Op {
+	case structs.IntentionOpCreate:
+		if err := s.prepareApplyCreate(authz, &entMeta, args); err != nil {
 			return err
 		}
+	case structs.IntentionOpUpdate:
+		if err := s.prepareApplyUpdate(authz, &entMeta, args); err != nil {
+			return err
+		}
+	case structs.IntentionOpDelete:
+		if err := s.prepareApplyDelete(authz, &entMeta, args); err != nil {
+			return err
+		}
+	default:
+		return fmt.Errorf("Invalid Intention operation: %v", args.Op)
 	}

-	// make sure we set the hash prior to raft application
-	args.Intention.SetHash(true)
+	// setup the reply which will have been filled in by one of the 3 preparedApply* funcs
+	*reply = args.Intention.ID

 	// Commit
 	resp, err := s.srv.raftApply(structs.IntentionRequestType, args)
@ -240,10 +300,18 @@ func (s *Intention) Match(
 	}

 	if rule != nil {
-		// We go through each entry and test the destination to check if it
-		// matches.
+		var authzContext acl.AuthorizerContext
+		// Go through each entry to ensure we have intention:read for the resource.
+
+		// TODO - should we do this instead of filtering the result set? This will only allow
+		// queries for which the token has intention:read permissions on the requested side
+		// of the service. Should it instead return all matches that it would be able to list.
+		// if so we should remove this and call filterACL instead. Based on how this is used
+		// its probably fine. If you have intention read on the source just do a source type
+		// matching, if you have it on the dest then perform a dest type match.
 		for _, entry := range args.Match.Entries {
-			if prefix := entry.Name; prefix != "" && rule.IntentionRead(prefix, nil) != acl.Allow {
+			entry.FillAuthzContext(&authzContext)
+			if prefix := entry.Name; prefix != "" && rule.IntentionRead(prefix, &authzContext) != acl.Allow {
 				s.srv.logger.Printf("[WARN] consul.intention: Operation on intention prefix '%s' denied due to ACLs", prefix)
 				return acl.ErrPermissionDenied
 			}
@ -307,9 +375,14 @@ func (s *Intention) Check(

 	// Perform the ACL check. For Check we only require ServiceRead and
 	// NOT IntentionRead because the Check API only returns pass/fail and
-	// returns no other information about the intentions used.
+	// returns no other information about the intentions used. We could check
+	// both the source and dest side but only checking dest also has the nice
+	// benefit of only returning a passing status if the token would be able
+	// to discover the dest service and connect to it.
 	if prefix, ok := query.GetACLPrefix(); ok {
-		if rule != nil && rule.ServiceRead(prefix, nil) != acl.Allow {
+		var authzContext acl.AuthorizerContext
+		query.FillAuthzContext(&authzContext)
+		if rule != nil && rule.ServiceRead(prefix, &authzContext) != acl.Allow {
 			s.srv.logger.Printf("[WARN] consul.intention: test on intention '%s' denied due to ACLs", prefix)
 			return acl.ErrPermissionDenied
 		}
--- a/agent/consul/intention_endpoint_test.go
+++ b/agent/consul/intention_endpoint_test.go
@ -8,7 +8,7 @@ import (
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/testrpc"
-	"github.com/hashicorp/net-rpc-msgpackrpc"
+	msgpackrpc "github.com/hashicorp/net-rpc-msgpackrpc"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@ -389,6 +389,325 @@ service "foo" {
 	}
 }

+func TestIntention_WildcardACLEnforcement(t *testing.T) {
+	t.Parallel()
+
+	dir, srv := testACLServerWithConfig(t, nil, false)
+	defer os.RemoveAll(dir)
+	defer srv.Shutdown()
+	codec := rpcClient(t, srv)
+	defer codec.Close()
+
+	testrpc.WaitForLeader(t, srv.RPC, "dc1")
+
+	// create some test policies.
+
+	writeToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `service_prefix "" { policy = "deny" intentions = "write" }`)
+	require.NoError(t, err)
+	readToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `service_prefix "" { policy = "deny" intentions = "read" }`)
+	require.NoError(t, err)
+	exactToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `service "*" { policy = "deny" intentions = "write" }`)
+	require.NoError(t, err)
+	wildcardPrefixToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `service_prefix "*" { policy = "deny" intentions = "write" }`)
+	require.NoError(t, err)
+	fooToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `service "foo" { policy = "deny" intentions = "write" }`)
+	require.NoError(t, err)
+	denyToken, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `service_prefix "" { policy = "deny" intentions = "deny" }`)
+	require.NoError(t, err)
+
+	doIntentionCreate := func(t *testing.T, token string, deny bool) string {
+		t.Helper()
+		ixn := structs.IntentionRequest{
+			Datacenter: "dc1",
+			Op:         structs.IntentionOpCreate,
+			Intention: &structs.Intention{
+				SourceNS:        "default",
+				SourceName:      "*",
+				DestinationNS:   "default",
+				DestinationName: "*",
+				Action:          structs.IntentionActionAllow,
+				SourceType:      structs.IntentionSourceConsul,
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		}
+		var reply string
+		err := msgpackrpc.CallWithCodec(codec, "Intention.Apply", &ixn, &reply)
+		if deny {
+			require.Error(t, err)
+			require.True(t, acl.IsErrPermissionDenied(err))
+			return ""
+		} else {
+			require.NoError(t, err)
+			require.NotEmpty(t, reply)
+			return reply
+		}
+	}
+
+	t.Run("deny-write-for-read-token", func(t *testing.T) {
+		// This tests ensures that tokens with only read access to all intentions
+		// cannot create a wildcard intention
+		doIntentionCreate(t, readToken.SecretID, true)
+	})
+
+	t.Run("deny-write-for-exact-wildcard-rule", func(t *testing.T) {
+		// This test ensures that having a rules like:
+		// service "*" {
+		//    intentions = "write"
+		// }
+		// will not actually allow creating an intention with a wildcard service name
+		doIntentionCreate(t, exactToken.SecretID, true)
+	})
+
+	t.Run("deny-write-for-prefix-wildcard-rule", func(t *testing.T) {
+		// This test ensures that having a rules like:
+		// service_prefix "*" {
+		//    intentions = "write"
+		// }
+		// will not actually allow creating an intention with a wildcard service name
+		doIntentionCreate(t, wildcardPrefixToken.SecretID, true)
+	})
+
+	var intentionID string
+	allowWriteOk := t.Run("allow-write", func(t *testing.T) {
+		// tests that a token with all the required privileges can create
+		// intentions with a wildcard destination
+		intentionID = doIntentionCreate(t, writeToken.SecretID, false)
+	})
+
+	requireAllowWrite := func(t *testing.T) {
+		t.Helper()
+		if !allowWriteOk {
+			t.Skip("Skipping because the allow-write subtest failed")
+		}
+	}
+
+	doIntentionRead := func(t *testing.T, token string, deny bool) {
+		t.Helper()
+		requireAllowWrite(t)
+		req := &structs.IntentionQueryRequest{
+			Datacenter:   "dc1",
+			IntentionID:  intentionID,
+			QueryOptions: structs.QueryOptions{Token: token},
+		}
+
+		var resp structs.IndexedIntentions
+		err := msgpackrpc.CallWithCodec(codec, "Intention.Get", req, &resp)
+		if deny {
+			require.Error(t, err)
+			require.True(t, acl.IsErrPermissionDenied(err))
+		} else {
+			require.NoError(t, err)
+			require.Len(t, resp.Intentions, 1)
+			require.Equal(t, "*", resp.Intentions[0].DestinationName)
+		}
+	}
+
+	t.Run("allow-read-for-write-token", func(t *testing.T) {
+		doIntentionRead(t, writeToken.SecretID, false)
+	})
+
+	t.Run("allow-read-for-read-token", func(t *testing.T) {
+		doIntentionRead(t, readToken.SecretID, false)
+	})
+
+	t.Run("allow-read-for-exact-wildcard-token", func(t *testing.T) {
+		// this is allowed because, the effect of the policy is to grant
+		// intention:write on the service named "*". When reading the
+		// intention we will validate that the token has read permissions
+		// for any intention that would match the wildcard.
+		doIntentionRead(t, exactToken.SecretID, false)
+	})
+
+	t.Run("allow-read-for-prefix-wildcard-token", func(t *testing.T) {
+		// this is allowed for the same reasons as for the
+		// exact-wildcard-token case
+		doIntentionRead(t, wildcardPrefixToken.SecretID, false)
+	})
+
+	t.Run("deny-read-for-deny-token", func(t *testing.T) {
+		doIntentionRead(t, denyToken.SecretID, true)
+	})
+
+	doIntentionList := func(t *testing.T, token string, deny bool) {
+		t.Helper()
+		requireAllowWrite(t)
+		req := &structs.DCSpecificRequest{
+			Datacenter:   "dc1",
+			QueryOptions: structs.QueryOptions{Token: token},
+		}
+
+		var resp structs.IndexedIntentions
+		err := msgpackrpc.CallWithCodec(codec, "Intention.List", req, &resp)
+		// even with permission denied this should return success but with an empty list
+		require.NoError(t, err)
+		if deny {
+			require.Empty(t, resp.Intentions)
+		} else {
+			require.Len(t, resp.Intentions, 1)
+			require.Equal(t, "*", resp.Intentions[0].DestinationName)
+		}
+	}
+
+	t.Run("allow-list-for-write-token", func(t *testing.T) {
+		doIntentionList(t, writeToken.SecretID, false)
+	})
+
+	t.Run("allow-list-for-read-token", func(t *testing.T) {
+		doIntentionList(t, readToken.SecretID, false)
+	})
+
+	t.Run("allow-list-for-exact-wildcard-token", func(t *testing.T) {
+		doIntentionList(t, exactToken.SecretID, false)
+	})
+
+	t.Run("allow-list-for-prefix-wildcard-token", func(t *testing.T) {
+		doIntentionList(t, wildcardPrefixToken.SecretID, false)
+	})
+
+	t.Run("deny-list-for-deny-token", func(t *testing.T) {
+		doIntentionList(t, denyToken.SecretID, true)
+	})
+
+	doIntentionMatch := func(t *testing.T, token string, deny bool) {
+		t.Helper()
+		requireAllowWrite(t)
+		req := &structs.IntentionQueryRequest{
+			Datacenter: "dc1",
+			Match: &structs.IntentionQueryMatch{
+				Type: structs.IntentionMatchDestination,
+				Entries: []structs.IntentionMatchEntry{
+					structs.IntentionMatchEntry{
+						Namespace: "default",
+						Name:      "*",
+					},
+				},
+			},
+			QueryOptions: structs.QueryOptions{Token: token},
+		}
+
+		var resp structs.IndexedIntentionMatches
+		err := msgpackrpc.CallWithCodec(codec, "Intention.Match", req, &resp)
+		if deny {
+			require.Error(t, err)
+			require.Empty(t, resp.Matches)
+		} else {
+			require.NoError(t, err)
+			require.Len(t, resp.Matches, 1)
+			require.Len(t, resp.Matches[0], 1)
+			require.Equal(t, "*", resp.Matches[0][0].DestinationName)
+		}
+	}
+
+	t.Run("allow-match-for-write-token", func(t *testing.T) {
+		doIntentionMatch(t, writeToken.SecretID, false)
+	})
+
+	t.Run("allow-match-for-read-token", func(t *testing.T) {
+		doIntentionMatch(t, readToken.SecretID, false)
+	})
+
+	t.Run("allow-match-for-exact-wildcard-token", func(t *testing.T) {
+		doIntentionMatch(t, exactToken.SecretID, false)
+	})
+
+	t.Run("allow-match-for-prefix-wildcard-token", func(t *testing.T) {
+		doIntentionMatch(t, wildcardPrefixToken.SecretID, false)
+	})
+
+	t.Run("deny-match-for-deny-token", func(t *testing.T) {
+		doIntentionMatch(t, denyToken.SecretID, true)
+	})
+
+	doIntentionUpdate := func(t *testing.T, token string, dest string, deny bool) {
+		t.Helper()
+		requireAllowWrite(t)
+		ixn := structs.IntentionRequest{
+			Datacenter: "dc1",
+			Op:         structs.IntentionOpUpdate,
+			Intention: &structs.Intention{
+				ID:              intentionID,
+				SourceNS:        "default",
+				SourceName:      "*",
+				DestinationNS:   "default",
+				DestinationName: dest,
+				Action:          structs.IntentionActionAllow,
+				SourceType:      structs.IntentionSourceConsul,
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		}
+		var reply string
+		err := msgpackrpc.CallWithCodec(codec, "Intention.Apply", &ixn, &reply)
+		if deny {
+			require.Error(t, err)
+			require.True(t, acl.IsErrPermissionDenied(err))
+		} else {
+			require.NoError(t, err)
+		}
+	}
+
+	t.Run("deny-update-for-foo-token", func(t *testing.T) {
+		doIntentionUpdate(t, fooToken.SecretID, "foo", true)
+	})
+
+	t.Run("allow-update-for-prefix-token", func(t *testing.T) {
+		// this tests that regardless of going from a wildcard intention
+		// to a non-wildcard or the opposite direction that the permissions
+		// are checked correctly. This also happens to leave the intention
+		// in a state ready for verifying similar things with deletion
+		doIntentionUpdate(t, writeToken.SecretID, "foo", false)
+		doIntentionUpdate(t, writeToken.SecretID, "*", false)
+	})
+
+	doIntentionDelete := func(t *testing.T, token string, deny bool) {
+		t.Helper()
+		requireAllowWrite(t)
+		ixn := structs.IntentionRequest{
+			Datacenter: "dc1",
+			Op:         structs.IntentionOpDelete,
+			Intention: &structs.Intention{
+				ID: intentionID,
+			},
+			WriteRequest: structs.WriteRequest{Token: token},
+		}
+		var reply string
+		err := msgpackrpc.CallWithCodec(codec, "Intention.Apply", &ixn, &reply)
+		if deny {
+			require.Error(t, err)
+			require.True(t, acl.IsErrPermissionDenied(err))
+		} else {
+			require.NoError(t, err)
+		}
+	}
+
+	t.Run("deny-delete-for-read-token", func(t *testing.T) {
+		doIntentionDelete(t, readToken.SecretID, true)
+	})
+
+	t.Run("deny-delete-for-exact-wildcard-rule", func(t *testing.T) {
+		// This test ensures that having a rules like:
+		// service "*" {
+		//    intentions = "write"
+		// }
+		// will not actually allow deleting an intention with a wildcard service name
+		doIntentionDelete(t, exactToken.SecretID, true)
+	})
+
+	t.Run("deny-delete-for-prefix-wildcard-rule", func(t *testing.T) {
+		// This test ensures that having a rules like:
+		// service_prefix "*" {
+		//    intentions = "write"
+		// }
+		// will not actually allow creating an intention with a wildcard service name
+		doIntentionDelete(t, wildcardPrefixToken.SecretID, true)
+	})
+
+	t.Run("allow-delete", func(t *testing.T) {
+		// tests that a token with all the required privileges can delete
+		// intentions with a wildcard destination
+		doIntentionDelete(t, writeToken.SecretID, false)
+	})
+}
+
 // Test apply with delete and a default deny ACL
 func TestIntentionApply_aclDelete(t *testing.T) {
 	t.Parallel()
@ -1182,13 +1501,7 @@ service "bar" {
 func TestIntentionCheck_match(t *testing.T) {
 	t.Parallel()

-	require := require.New(t)
-	dir1, s1 := testServerWithConfig(t, func(c *Config) {
-		c.ACLDatacenter = "dc1"
-		c.ACLsEnabled = true
-		c.ACLMasterToken = "root"
-		c.ACLDefaultPolicy = "deny"
-	})
+	dir1, s1 := testACLServerWithConfig(t, nil, false)
 	defer os.RemoveAll(dir1)
 	defer s1.Shutdown()
 	codec := rpcClient(t, s1)
@ -1196,33 +1509,15 @@ func TestIntentionCheck_match(t *testing.T) {

 	testrpc.WaitForLeader(t, s1.RPC, "dc1")

-	// Create an ACL with service read permissions. This will grant permission.
-	var token string
-	{
-		var rules = `
-service "bar" {
-	policy = "read"
-}`
-
-		req := structs.ACLRequest{
-			Datacenter: "dc1",
-			Op:         structs.ACLSet,
-			ACL: structs.ACL{
-				Name:  "User token",
-				Type:  structs.ACLTokenTypeClient,
-				Rules: rules,
-			},
-			WriteRequest: structs.WriteRequest{Token: "root"},
-		}
-		require.Nil(msgpackrpc.CallWithCodec(codec, "ACL.Apply", &req, &token))
-	}
+	token, err := upsertTestTokenWithPolicyRules(codec, TestDefaultMasterToken, "dc1", `service "api" { policy = "read" }`)
+	require.NoError(t, err)

 	// Create some intentions
 	{
 		insert := [][]string{
-			{"foo", "*", "foo", "*"},
-			{"foo", "*", "foo", "bar"},
-			{"bar", "*", "foo", "bar"}, // duplicate destination different source
+			{"web", "db"},
+			{"api", "db"},
+			{"web", "api"},
 		}

 		for _, v := range insert {
@ -1230,18 +1525,17 @@ service "bar" {
 				Datacenter: "dc1",
 				Op:         structs.IntentionOpCreate,
 				Intention: &structs.Intention{
-					SourceNS:        v[0],
-					SourceName:      v[1],
-					DestinationNS:   v[2],
-					DestinationName: v[3],
+					SourceNS:        "default",
+					SourceName:      v[0],
+					DestinationNS:   "default",
+					DestinationName: v[1],
 					Action:          structs.IntentionActionAllow,
 				},
+				WriteRequest: structs.WriteRequest{Token: TestDefaultMasterToken},
 			}
-			ixn.WriteRequest.Token = "root"
-
 			// Create
 			var reply string
-			require.Nil(msgpackrpc.CallWithCodec(codec, "Intention.Apply", &ixn, &reply))
+			require.NoError(t, msgpackrpc.CallWithCodec(codec, "Intention.Apply", &ixn, &reply))
 		}
 	}

@ -1249,33 +1543,33 @@ service "bar" {
 	req := &structs.IntentionQueryRequest{
 		Datacenter: "dc1",
 		Check: &structs.IntentionQueryCheck{
-			SourceNS:        "foo",
-			SourceName:      "qux",
-			DestinationNS:   "foo",
-			DestinationName: "bar",
+			SourceNS:        "default",
+			SourceName:      "web",
+			DestinationNS:   "default",
+			DestinationName: "api",
 			SourceType:      structs.IntentionSourceConsul,
 		},
+		QueryOptions: structs.QueryOptions{Token: token.SecretID},
 	}
-	req.Token = token
 	var resp structs.IntentionQueryCheckResponse
-	require.Nil(msgpackrpc.CallWithCodec(codec, "Intention.Check", req, &resp))
-	require.True(resp.Allowed)
+	require.NoError(t, msgpackrpc.CallWithCodec(codec, "Intention.Check", req, &resp))
+	require.True(t, resp.Allowed)

 	// Test no match for sanity
 	{
 		req := &structs.IntentionQueryRequest{
 			Datacenter: "dc1",
 			Check: &structs.IntentionQueryCheck{
-				SourceNS:        "baz",
-				SourceName:      "qux",
-				DestinationNS:   "foo",
-				DestinationName: "bar",
+				SourceNS:        "default",
+				SourceName:      "db",
+				DestinationNS:   "default",
+				DestinationName: "api",
 				SourceType:      structs.IntentionSourceConsul,
 			},
+			QueryOptions: structs.QueryOptions{Token: token.SecretID},
 		}
-		req.Token = token
 		var resp structs.IntentionQueryCheckResponse
-		require.Nil(msgpackrpc.CallWithCodec(codec, "Intention.Check", req, &resp))
-		require.False(resp.Allowed)
+		require.NoError(t, msgpackrpc.CallWithCodec(codec, "Intention.Check", req, &resp))
+		require.False(t, resp.Allowed)
 	}
 }
--- a/agent/consul/leader_connect_test.go
+++ b/agent/consul/leader_connect_test.go
@ -668,11 +668,12 @@ func TestLeader_ReplicateIntentions(t *testing.T) {

 	s1.tokens.UpdateAgentToken("root", tokenStore.TokenSourceConfig)

+	replicationRules := `acl = "read" service_prefix "" { policy = "read" intentions = "read" } operator = "write" `
 	// create some tokens
-	replToken1, err := upsertTestTokenWithPolicyRules(codec, "root", "dc1", `acl = "read" operator = "write"`)
+	replToken1, err := upsertTestTokenWithPolicyRules(codec, "root", "dc1", replicationRules)
 	require.NoError(err)

-	replToken2, err := upsertTestTokenWithPolicyRules(codec, "root", "dc1", `acl = "read" operator = "write"`)
+	replToken2, err := upsertTestTokenWithPolicyRules(codec, "root", "dc1", replicationRules)
 	require.NoError(err)

 	// dc2 as a secondary DC
--- a/agent/consul/server.go
+++ b/agent/consul/server.go
@ -108,9 +108,8 @@ var (
 // Server is Consul server which manages the service discovery,
 // health checking, DC forwarding, Raft, and multiple Serf pools.
 type Server struct {
-	// enterpriseACLConfig is the Consul Enterprise specific items
-	// necessary for ACLs
-	enterpriseACLConfig *acl.Config
+	// aclConfig is the configuration for the ACL system
+	aclConfig *acl.Config

 	// acls is used to resolve tokens to effective policies
 	acls *ACLResolver
@ -397,15 +396,15 @@ func NewServerLogger(config *Config, logger *log.Logger, tokens *token.Store, tl
 	// Initialize the stats fetcher that autopilot will use.
 	s.statsFetcher = NewStatsFetcher(logger, s.connPool, s.config.Datacenter)

-	s.enterpriseACLConfig = newEnterpriseACLConfig(logger)
+	s.aclConfig = newACLConfig(logger)
 	s.useNewACLs = 0
 	aclConfig := ACLResolverConfig{
-		Config:           config,
-		Delegate:         s,
-		CacheConfig:      serverACLCacheConfig,
-		AutoDisable:      false,
-		Logger:           logger,
-		EnterpriseConfig: s.enterpriseACLConfig,
+		Config:      config,
+		Delegate:    s,
+		CacheConfig: serverACLCacheConfig,
+		AutoDisable: false,
+		Logger:      logger,
+		ACLConfig:   s.aclConfig,
 	}
 	// Initialize the ACL resolver.
 	if s.acls, err = NewACLResolver(&aclConfig); err != nil {
--- a/agent/consul/server_test.go
+++ b/agent/consul/server_test.go
@ -29,6 +29,27 @@ import (
 	"github.com/stretchr/testify/require"
 )

+const (
+	TestDefaultMasterToken = "d9f05e83-a7ae-47ce-839e-c0d53a68c00a"
+)
+
+// testServerACLConfig wraps another arbitrary Config altering callback
+// to setup some common ACL configurations. A new callback func will
+// be returned that has the original callback invoked after setting
+// up all of the ACL configurations (so they can still be overridden)
+func testServerACLConfig(cb func(*Config)) func(*Config) {
+	return func(c *Config) {
+		c.ACLDatacenter = "dc1"
+		c.ACLsEnabled = true
+		c.ACLMasterToken = TestDefaultMasterToken
+		c.ACLDefaultPolicy = "deny"
+
+		if cb != nil {
+			cb(c)
+		}
+	}
+}
+
 func configureTLS(config *Config) {
 	config.CAFile = "../../test/ca/root.cer"
 	config.CertFile = "../../test/key/ourdomain.cer"
@ -207,6 +228,17 @@ func testServerWithConfig(t *testing.T, cb func(*Config)) (string, *Server) {
 	return dir, srv
 }

+// cb is a function that can alter the test servers configuration prior to the server starting.
+func testACLServerWithConfig(t *testing.T, cb func(*Config), initReplicationToken bool) (string, *Server) {
+	dir, srv := testServerWithConfig(t, testServerACLConfig(cb))
+
+	if initReplicationToken {
+		// setup some tokens here so we get less warnings in the logs
+		srv.tokens.UpdateReplicationToken(TestDefaultMasterToken, token.TokenSourceConfig)
+	}
+	return dir, srv
+}
+
 func newServer(c *Config) (*Server, error) {
 	// chain server up notification
 	oldNotify := c.NotifyListen
--- a/agent/consul/state/intention.go
+++ b/agent/consul/state/intention.go
@ -349,14 +349,14 @@ func (s *Store) intentionMatchGetParams(entry structs.IntentionMatchEntry) ([][]
 	// We always query for "*/*" so include that. If the namespace is a
 	// wildcard, then we're actually done.
 	result := make([][]interface{}, 0, 3)
-	result = append(result, []interface{}{"*", "*"})
-	if entry.Namespace == structs.IntentionWildcard {
+	result = append(result, []interface{}{structs.WildcardSpecifier, structs.WildcardSpecifier})
+	if entry.Namespace == structs.WildcardSpecifier {
 		return result, nil
 	}

 	// Search for NS/* intentions. If we have a wildcard name, then we're done.
-	result = append(result, []interface{}{entry.Namespace, "*"})
-	if entry.Name == structs.IntentionWildcard {
+	result = append(result, []interface{}{entry.Namespace, structs.WildcardSpecifier})
+	if entry.Name == structs.WildcardSpecifier {
 		return result, nil
 	}

--- a/agent/structs/connect.go
+++ b/agent/structs/connect.go
@ -6,6 +6,9 @@ type ConnectAuthorizeRequest struct {
 	// Target is the name of the service that is being requested.
 	Target string

+	// EnterpriseMeta is the embedded Consul Enterprise specific metadata
+	EnterpriseMeta
+
 	// ClientCertURI is a unique identifier for the requesting client. This
 	// is currently the URI SAN from the TLS client certificate.
 	//
--- a/agent/structs/connect_oss.go
+++ b/agent/structs/connect_oss.go
@ -0,0 +1,7 @@
+// +build !consulent
+
+package structs
+
+func (req *ConnectAuthorizeRequest) TargetNamespace() string {
+	return IntentionDefaultNamespace
+}
--- a/agent/structs/intention.go
+++ b/agent/structs/intention.go
@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"

+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/cache"
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/go-multierror"
@ -17,9 +18,6 @@ import (
 )

 const (
-	// IntentionWildcard is the wildcard value.
-	IntentionWildcard = "*"
-
 	// IntentionDefaultNamespace is the default namespace value.
 	// NOTE(mitchellh): This is only meant to be a temporary constant.
 	// When namespaces are introduced, we should delete this constant and
@ -175,36 +173,36 @@ func (x *Intention) Validate() error {
 	}

 	// Wildcard usage verification
-	if x.SourceNS != IntentionWildcard {
-		if strings.Contains(x.SourceNS, IntentionWildcard) {
+	if x.SourceNS != WildcardSpecifier {
+		if strings.Contains(x.SourceNS, WildcardSpecifier) {
 			result = multierror.Append(result, fmt.Errorf(
 				"SourceNS: wildcard character '*' cannot be used with partial values"))
 		}
 	}
-	if x.SourceName != IntentionWildcard {
-		if strings.Contains(x.SourceName, IntentionWildcard) {
+	if x.SourceName != WildcardSpecifier {
+		if strings.Contains(x.SourceName, WildcardSpecifier) {
 			result = multierror.Append(result, fmt.Errorf(
 				"SourceName: wildcard character '*' cannot be used with partial values"))
 		}

-		if x.SourceNS == IntentionWildcard {
+		if x.SourceNS == WildcardSpecifier {
 			result = multierror.Append(result, fmt.Errorf(
 				"SourceName: exact value cannot follow wildcard namespace"))
 		}
 	}
-	if x.DestinationNS != IntentionWildcard {
-		if strings.Contains(x.DestinationNS, IntentionWildcard) {
+	if x.DestinationNS != WildcardSpecifier {
+		if strings.Contains(x.DestinationNS, WildcardSpecifier) {
 			result = multierror.Append(result, fmt.Errorf(
 				"DestinationNS: wildcard character '*' cannot be used with partial values"))
 		}
 	}
-	if x.DestinationName != IntentionWildcard {
-		if strings.Contains(x.DestinationName, IntentionWildcard) {
+	if x.DestinationName != WildcardSpecifier {
+		if strings.Contains(x.DestinationName, WildcardSpecifier) {
 			result = multierror.Append(result, fmt.Errorf(
 				"DestinationName: wildcard character '*' cannot be used with partial values"))
 		}

-		if x.DestinationNS == IntentionWildcard {
+		if x.DestinationNS == WildcardSpecifier {
 			result = multierror.Append(result, fmt.Errorf(
 				"DestinationName: exact value cannot follow wildcard namespace"))
 		}
@ -247,6 +245,43 @@ func (x *Intention) Validate() error {
 	return result
 }

+func (ixn *Intention) CanRead(authz acl.Authorizer) bool {
+	if authz == nil {
+		return true
+	}
+	var authzContext acl.AuthorizerContext
+
+	if ixn.SourceName != "" {
+		ixn.FillAuthzContext(&authzContext, false)
+		if authz.IntentionRead(ixn.SourceName, &authzContext) == acl.Allow {
+			return true
+		}
+	}
+
+	if ixn.DestinationName != "" {
+		ixn.FillAuthzContext(&authzContext, true)
+		if authz.IntentionRead(ixn.DestinationName, &authzContext) == acl.Allow {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (ixn *Intention) CanWrite(authz acl.Authorizer) bool {
+	if authz == nil {
+		return true
+	}
+	var authzContext acl.AuthorizerContext
+
+	if ixn.DestinationName == "" {
+		return false
+	}
+
+	ixn.FillAuthzContext(&authzContext, true)
+	return authz.IntentionWrite(ixn.DestinationName, &authzContext) == acl.Allow
+}
+
 // UpdatePrecedence sets the Precedence value based on the fields of this
 // structure.
 func (x *Intention) UpdatePrecedence() {
@ -276,27 +311,20 @@ func (x *Intention) UpdatePrecedence() {
 // the given namespace and name.
 func (x *Intention) countExact(ns, n string) int {
 	// If NS is wildcard, it must be zero since wildcards only follow exact
-	if ns == IntentionWildcard {
+	if ns == WildcardSpecifier {
 		return 0
 	}

 	// Same reasoning as above, a wildcard can only follow an exact value
 	// and an exact value cannot follow a wildcard, so if name is a wildcard
 	// we must have exactly one.
-	if n == IntentionWildcard {
+	if n == WildcardSpecifier {
 		return 1
 	}

 	return 2
 }

-// GetACLPrefix returns the prefix to look up the ACL policy for this
-// intention, and a boolean noting whether the prefix is valid to check
-// or not. You must check the ok value before using the prefix.
-func (x *Intention) GetACLPrefix() (string, bool) {
-	return x.DestinationName, x.DestinationName != ""
-}
-
 // String returns a human-friendly string for this intention.
 func (x *Intention) String() string {
 	return fmt.Sprintf("%s %s/%s => %s/%s (ID: %s, Precedence: %d)",
--- a/agent/structs/intention_oss.go
+++ b/agent/structs/intention_oss.go
@ -0,0 +1,40 @@
+// +build !consulent
+
+package structs
+
+import (
+	"github.com/hashicorp/consul/acl"
+)
+
+// FillAuthzContext can fill in an acl.AuthorizerContext object to setup
+// extra parameters for ACL enforcement. In OSS there is currently nothing
+// extra to be done.
+func (_ *Intention) FillAuthzContext(_ *acl.AuthorizerContext, _ bool) {
+	// do nothing
+}
+
+// FillAuthzContext can fill in an acl.AuthorizerContext object to setup
+// extra parameters for ACL enforcement. In OSS there is currently nothing
+// extra to be done.
+func (_ *IntentionMatchEntry) FillAuthzContext(_ *acl.AuthorizerContext) {
+	// do nothing
+}
+
+// FillAuthzContext can fill in an acl.AuthorizerContext object to setup
+// extra parameters for ACL enforcement. In OSS there is currently nothing
+// extra to be done.
+func (_ *IntentionQueryCheck) FillAuthzContext(_ *acl.AuthorizerContext) {
+	// do nothing
+}
+
+// DefaultNamespaces will populate both the SourceNS and DestinationNS fields
+// if they are empty with the proper defaults.
+func (ixn *Intention) DefaultNamespaces(_ *EnterpriseMeta) {
+	// Until we support namespaces, we force all namespaces to be default
+	if ixn.SourceNS == "" {
+		ixn.SourceNS = IntentionDefaultNamespace
+	}
+	if ixn.DestinationNS == "" {
+		ixn.DestinationNS = IntentionDefaultNamespace
+	}
+}
--- a/agent/structs/intention_test.go
+++ b/agent/structs/intention_test.go
@ -5,42 +5,123 @@ import (
 	"strings"
 	"testing"

+	"github.com/hashicorp/consul/acl"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )

-func TestIntentionGetACLPrefix(t *testing.T) {
-	cases := []struct {
-		Name     string
-		Input    *Intention
-		Expected string
-	}{
-		{
-			"unset name",
-			&Intention{DestinationName: ""},
-			"",
-		},
+func TestIntention_ACLs(t *testing.T) {
+	t.Parallel()
+	type testCase struct {
+		intention Intention
+		rules     string
+		read      bool
+		write     bool
+	}

-		{
-			"set name",
-			&Intention{DestinationName: "fo"},
-			"fo",
+	cases := map[string]testCase{
+		"all-denied": testCase{
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "web",
+				DestinationNS:   "default",
+				DestinationName: "api",
+			},
+			read:  false,
+			write: false,
+		},
+		"deny-write-read-dest": testCase{
+			rules: `service "api" { policy = "deny" intentions = "read" }`,
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "web",
+				DestinationNS:   "default",
+				DestinationName: "api",
+			},
+			read:  true,
+			write: false,
+		},
+		"deny-write-read-source": testCase{
+			rules: `service "web" { policy = "deny" intentions = "read" }`,
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "web",
+				DestinationNS:   "default",
+				DestinationName: "api",
+			},
+			read:  true,
+			write: false,
+		},
+		"allow-write-with-dest-write": testCase{
+			rules: `service "api" { policy = "deny" intentions = "write" }`,
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "web",
+				DestinationNS:   "default",
+				DestinationName: "api",
+			},
+			read:  true,
+			write: true,
+		},
+		"deny-write-with-source-write": testCase{
+			rules: `service "web" { policy = "deny" intentions = "write" }`,
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "web",
+				DestinationNS:   "default",
+				DestinationName: "api",
+			},
+			read:  true,
+			write: false,
+		},
+		"deny-wildcard-write-allow-read": testCase{
+			rules: `service "*" { policy = "deny" intentions = "write" }`,
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "*",
+				DestinationNS:   "default",
+				DestinationName: "*",
+			},
+			// technically having been granted read/write on any intention will allow
+			// read access for this rule
+			read:  true,
+			write: false,
+		},
+		"allow-wildcard-write": testCase{
+			rules: `service_prefix "" { policy = "deny" intentions = "write" }`,
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "*",
+				DestinationNS:   "default",
+				DestinationName: "*",
+			},
+			read:  true,
+			write: true,
+		},
+		"allow-wildcard-read": testCase{
+			rules: `service "foo" { policy = "deny" intentions = "read" }`,
+			intention: Intention{
+				SourceNS:        "default",
+				SourceName:      "*",
+				DestinationNS:   "default",
+				DestinationName: "*",
+			},
+			read:  true,
+			write: false,
 		},
 	}

-	for _, tc := range cases {
-		t.Run(tc.Name, func(t *testing.T) {
-			actual, ok := tc.Input.GetACLPrefix()
-			if tc.Expected == "" {
-				if !ok {
-					return
-				}
+	config := acl.Config{
+		WildcardName: WildcardSpecifier,
+	}

-				t.Fatal("should not be ok")
-			}
+	for name, tcase := range cases {
+		t.Run(name, func(t *testing.T) {
+			authz, err := acl.NewAuthorizerFromRules("", 0, tcase.rules, acl.SyntaxCurrent, &config, nil)
+			require.NoError(t, err)

-			if actual != tc.Expected {
-				t.Fatalf("bad: %q", actual)
-			}
+			require.Equal(t, tcase.read, tcase.intention.CanRead(authz))
+			require.Equal(t, tcase.write, tcase.intention.CanWrite(authz))
 		})
 	}
 }
--- a/agent/structs/structs.go
+++ b/agent/structs/structs.go
@ -109,6 +109,10 @@ const (
 	// ends up being very small. If we see a value below this threshold,
 	// we multiply by time.Second
 	lockDelayMinThreshold = 1000
+
+	// WildcardSpecifier is the string which should be used for specifying a wildcard
+	// The exact semantics of the wildcard is left up to the code where its used.
+	WildcardSpecifier = "*"
 )

 var (
--- a/website/source/api/agent/connect.html.md
+++ b/website/source/api/agent/connect.html.md
@ -56,6 +56,12 @@ The table below shows this endpoint's support for
 - `ClientCertSerial` `(string: <required>)` - The colon-hex-encoded serial
  number for the requesting client cert. This is used to check against
  revocation lists.
+  
+- `Namespace` `(string: "")` - **(Enterprise Only)** Specifies the namespace of
+  the target service. If not provided in the JSON body, the value of
+  the `ns` URL query parameter or in the `X-Consul-Namespace` header will be used. 
+  If not provided at all, the namespace will be inherited from the request's ACL 
+  token or will default to the `default` namespace. Added in Consul 1.7.0.

 ### Sample Payload