diff --git a/acl/acl.go b/acl/acl.go
index 4ac88f01c9..49dc569b97 100644
--- a/acl/acl.go
+++ b/acl/acl.go
@@ -60,6 +60,10 @@ type ACL interface {
 	// EventWrite determines if a specific event may be fired.
 	EventWrite(string) bool
 
+	// IntentionDefault determines the default authorized behavior
+	// when no intentions match a Connect request.
+	IntentionDefault() bool
+
 	// IntentionRead determines if a specific intention can be read.
 	IntentionRead(string) bool
 
@@ -161,6 +165,10 @@ func (s *StaticACL) EventWrite(string) bool {
 	return s.defaultAllow
 }
 
+func (s *StaticACL) IntentionDefault() bool {
+	return s.defaultAllow
+}
+
 func (s *StaticACL) IntentionRead(string) bool {
 	return s.defaultAllow
 }
@@ -493,6 +501,13 @@ func (p *PolicyACL) EventWrite(name string) bool {
 	return p.parent.EventWrite(name)
 }
 
+// IntentionDefault returns whether the default behavior when there are
+// no matching intentions is to allow or deny.
+func (p *PolicyACL) IntentionDefault() bool {
+	// We always go up, this can't be determined by a policy.
+	return p.parent.IntentionDefault()
+}
+
 // IntentionRead checks if writing (creating, updating, or deleting) of an
 // intention is allowed.
 func (p *PolicyACL) IntentionRead(prefix string) bool {
diff --git a/acl/acl_test.go b/acl/acl_test.go
index 85f35f6066..263af0656d 100644
--- a/acl/acl_test.go
+++ b/acl/acl_test.go
@@ -53,6 +53,9 @@ func TestStaticACL(t *testing.T) {
 	if !all.EventWrite("foobar") {
 		t.Fatalf("should allow")
 	}
+	if !all.IntentionDefault() {
+		t.Fatalf("should allow")
+	}
 	if !all.IntentionWrite("foobar") {
 		t.Fatalf("should allow")
 	}
@@ -126,6 +129,9 @@ func TestStaticACL(t *testing.T) {
 	if none.EventWrite("") {
 		t.Fatalf("should not allow")
 	}
+	if none.IntentionDefault() {
+		t.Fatalf("should not allow")
+	}
 	if none.IntentionWrite("foo") {
 		t.Fatalf("should not allow")
 	}
@@ -193,6 +199,9 @@ func TestStaticACL(t *testing.T) {
 	if !manage.EventWrite("foobar") {
 		t.Fatalf("should allow")
 	}
+	if !manage.IntentionDefault() {
+		t.Fatalf("should allow")
+	}
 	if !manage.IntentionWrite("foobar") {
 		t.Fatalf("should allow")
 	}
@@ -454,6 +463,11 @@ func TestPolicyACL(t *testing.T) {
 			t.Fatalf("Prepared query fail: %#v", c)
 		}
 	}
+
+	// Check default intentions bubble up
+	if !acl.IntentionDefault() {
+		t.Fatal("should allow")
+	}
 }
 
 func TestPolicyACL_Parent(t *testing.T) {
@@ -607,6 +621,11 @@ func TestPolicyACL_Parent(t *testing.T) {
 	if acl.Snapshot() {
 		t.Fatalf("should not allow")
 	}
+
+	// Check default intentions
+	if acl.IntentionDefault() {
+		t.Fatal("should not allow")
+	}
 }
 
 func TestPolicyACL_Agent(t *testing.T) {
diff --git a/agent/agent_endpoint.go b/agent/agent_endpoint.go
index 02682e592d..5a9218c379 100644
--- a/agent/agent_endpoint.go
+++ b/agent/agent_endpoint.go
@@ -886,6 +886,10 @@ func (s *HTTPServer) AgentConnectCALeafCert(resp http.ResponseWriter, req *http.
 //
 // POST /v1/agent/connect/authorize
 func (s *HTTPServer) AgentConnectAuthorize(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+	// Fetch the token
+	var token string
+	s.parseToken(req, &token)
+
 	// Decode the request from the request body
 	var authReq structs.ConnectAuthorizeRequest
 	if err := decodeBody(req, &authReq, nil); err != nil {
@@ -925,7 +929,18 @@ func (s *HTTPServer) AgentConnectAuthorize(resp http.ResponseWriter, req *http.R
 		}, nil
 	}
 
-	// Get the intentions for this target service
+	// 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 := s.agent.resolveToken(token)
+	if err != nil {
+		return nil, err
+	}
+	if rule != nil && !rule.ServiceWrite(authReq.Target, nil) {
+		return nil, acl.ErrPermissionDenied
+	}
+
+	// Get the intentions for this target service.
 	args := &structs.IntentionQueryRequest{
 		Datacenter: s.agent.config.Datacenter,
 		Match: &structs.IntentionQueryMatch{
@@ -938,6 +953,7 @@ func (s *HTTPServer) AgentConnectAuthorize(resp http.ResponseWriter, req *http.R
 			},
 		},
 	}
+	args.Token = token
 	var reply structs.IndexedIntentionMatches
 	if err := s.agent.RPC("Intention.Match", args, &reply); err != nil {
 		return nil, err
@@ -956,15 +972,25 @@ func (s *HTTPServer) AgentConnectAuthorize(resp http.ResponseWriter, req *http.R
 		}
 	}
 
-	// If there was no matching intention, we always deny. Connect does
-	// support a blacklist (default allow) mode, but this works by appending
-	// */* => */* ALLOW intention to all Match requests. This means that
-	// the above should've matched. Therefore, if we reached here, something
-	// strange has happened and we should just deny the connection and err
-	// on the side of safety.
+	// No match, we need to determine the default behavior. We do this by
+	// specifying the anonymous token token, which will get that 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 = s.agent.resolveToken("")
+	if err != nil {
+		return nil, err
+	}
+	authz := true
+	reason := "ACLs disabled, access is allowed by default"
+	if rule != nil {
+		authz = rule.IntentionDefault()
+		reason = "Default behavior configured by ACLs"
+	}
+
 	return &connectAuthorizeResp{
-		Authorized: false,
-		Reason:     "No matching intention, denying",
+		Authorized: authz,
+		Reason:     reason,
 	}, nil
 }
 
diff --git a/agent/agent_endpoint_test.go b/agent/agent_endpoint_test.go
index cae7a4cccb..bc59f37002 100644
--- a/agent/agent_endpoint_test.go
+++ b/agent/agent_endpoint_test.go
@@ -2292,3 +2292,93 @@ func TestAgentConnectAuthorize_deny(t *testing.T) {
 	assert.False(obj.Authorized)
 	assert.Contains(obj.Reason, "Matched")
 }
+
+// Test that authorize fails without service:write for the target service.
+func TestAgentConnectAuthorize_serviceWrite(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	a := NewTestAgent(t.Name(), TestACLConfig())
+	defer a.Shutdown()
+
+	// Create an ACL
+	var token string
+	{
+		args := map[string]interface{}{
+			"Name":  "User Token",
+			"Type":  "client",
+			"Rules": `service "foo" { policy = "read" }`,
+		}
+		req, _ := http.NewRequest("PUT", "/v1/acl/create?token=root", jsonReader(args))
+		resp := httptest.NewRecorder()
+		obj, err := a.srv.ACLCreate(resp, req)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		aclResp := obj.(aclCreateResponse)
+		token = aclResp.ID
+	}
+
+	args := &structs.ConnectAuthorizeRequest{
+		Target:   "foo",
+		ClientID: connect.TestSpiffeIDService(t, "web").URI().String(),
+	}
+	req, _ := http.NewRequest("POST",
+		"/v1/agent/connect/authorize?token="+token, jsonReader(args))
+	resp := httptest.NewRecorder()
+	_, err := a.srv.AgentConnectAuthorize(resp, req)
+	assert.True(acl.IsErrPermissionDenied(err))
+}
+
+// Test when no intentions match w/ a default deny policy
+func TestAgentConnectAuthorize_defaultDeny(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	a := NewTestAgent(t.Name(), TestACLConfig())
+	defer a.Shutdown()
+
+	args := &structs.ConnectAuthorizeRequest{
+		Target:   "foo",
+		ClientID: connect.TestSpiffeIDService(t, "web").URI().String(),
+	}
+	req, _ := http.NewRequest("POST", "/v1/agent/connect/authorize?token=root", jsonReader(args))
+	resp := httptest.NewRecorder()
+	respRaw, err := a.srv.AgentConnectAuthorize(resp, req)
+	assert.Nil(err)
+	assert.Equal(200, resp.Code)
+
+	obj := respRaw.(*connectAuthorizeResp)
+	assert.False(obj.Authorized)
+	assert.Contains(obj.Reason, "Default behavior")
+}
+
+// Test when no intentions match w/ a default allow policy
+func TestAgentConnectAuthorize_defaultAllow(t *testing.T) {
+	t.Parallel()
+
+	assert := assert.New(t)
+	a := NewTestAgent(t.Name(), `
+		acl_datacenter = "dc1"
+		acl_default_policy = "allow"
+		acl_master_token = "root"
+		acl_agent_token = "root"
+		acl_agent_master_token = "towel"
+		acl_enforce_version_8 = true
+	`)
+	defer a.Shutdown()
+
+	args := &structs.ConnectAuthorizeRequest{
+		Target:   "foo",
+		ClientID: connect.TestSpiffeIDService(t, "web").URI().String(),
+	}
+	req, _ := http.NewRequest("POST", "/v1/agent/connect/authorize?token=root", jsonReader(args))
+	resp := httptest.NewRecorder()
+	respRaw, err := a.srv.AgentConnectAuthorize(resp, req)
+	assert.Nil(err)
+	assert.Equal(200, resp.Code)
+
+	obj := respRaw.(*connectAuthorizeResp)
+	assert.True(obj.Authorized)
+	assert.Contains(obj.Reason, "Default behavior")
+}