diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md
index 22ea407fa..3329d5ef9 100644
--- a/CHANGELOG_PENDING.md
+++ b/CHANGELOG_PENDING.md
@@ -1,8 +1,11 @@
 ### Improvements
 * Adds CI detector for Buildkite [#7933](https://github.com/pulumi/pulumi/pull/7933)
 
-- [CLI] Adding the ability to use `pulumi org set [name]` to set a default org
-  to use when creating a stacks in the Pulumi Service backend or Self -hosted Service
+- [cli] - Add `--exclude-protected` flag to `pulumi destroy`.
+  [#8359](https://github.com/pulumi/pulumi/pull/8359)
+
+- [cli] Adding the ability to use `pulumi org set [name]` to set a default org
+  to use when creating a stacks in the Pulumi Service backend or self-hosted Service
   [#8352](https://github.com/pulumi/pulumi/pull/8352)
 
 - [schema] Add IsOverlay option to disable codegen for particular types
diff --git a/pkg/cmd/pulumi/destroy.go b/pkg/cmd/pulumi/destroy.go
index a08555765..78b27bc35 100644
--- a/pkg/cmd/pulumi/destroy.go
+++ b/pkg/cmd/pulumi/destroy.go
@@ -1,4 +1,4 @@
-// Copyright 2016-2018, Pulumi Corporation.
+// Copyright 2016-2021, Pulumi Corporation.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -24,8 +24,10 @@ import (
 	"github.com/pulumi/pulumi/pkg/v3/backend"
 	"github.com/pulumi/pulumi/pkg/v3/backend/display"
 	"github.com/pulumi/pulumi/pkg/v3/engine"
+	"github.com/pulumi/pulumi/pkg/v3/resource/graph"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/util/cmdutil"
+	"github.com/pulumi/pulumi/sdk/v3/go/common/util/contract"
 	"github.com/pulumi/pulumi/sdk/v3/go/common/util/result"
 )
 
@@ -52,6 +54,7 @@ func newDestroyCmd() *cobra.Command {
 	var yes bool
 	var targets *[]string
 	var targetDependents bool
+	var excludeProtected bool
 
 	var cmd = &cobra.Command{
 		Use:        "destroy",
@@ -149,6 +152,29 @@ func newDestroyCmd() *cobra.Command {
 			if err != nil {
 				return result.FromError(err)
 			}
+
+			if targets != nil && len(*targets) > 0 && excludeProtected {
+				return result.FromError(errors.New("You cannot specify --target and --exclude-protected"))
+			}
+
+			var protectedCount int
+			if excludeProtected {
+				contract.Assert(len(targetUrns) == 0)
+				targetUrns, protectedCount, err = handleExcludeProtected(s)
+				if err != nil {
+					return result.FromError(err)
+				} else if protectedCount > 0 && len(targetUrns) == 0 {
+					if !jsonDisplay {
+						fmt.Printf("There were no unprotected resources to destroy. There are still %d"+
+							" protected resources associated with this stack.\n", protectedCount)
+					}
+					// We need to return now. Otherwise the update will conclude
+					// we tried to destroy everything and error for trying to
+					// destroy a protected resource.
+					return nil
+				}
+			}
+
 			opts.Engine = engine.UpdateOptions{
 				Parallel:                  parallel,
 				Debug:                     debug,
@@ -170,8 +196,10 @@ func newDestroyCmd() *cobra.Command {
 				SecretsManager:     sm,
 				Scopes:             cancellationScopes,
 			})
-
-			if res == nil && len(*targets) == 0 && !jsonDisplay {
+			if res == nil && protectedCount > 0 && !jsonDisplay {
+				fmt.Printf("All unprotected resources were destroyed. There are still %d protected resources"+
+					" associated with this stack.\n", protectedCount)
+			} else if res == nil && len(*targets) == 0 && !jsonDisplay {
 				fmt.Printf("The resources in the stack have been deleted, but the history and configuration "+
 					"associated with the stack are still maintained. \nIf you want to remove the stack "+
 					"completely, run 'pulumi stack rm %s'.\n", s.Ref())
@@ -202,6 +230,8 @@ func newDestroyCmd() *cobra.Command {
 	cmd.PersistentFlags().BoolVar(
 		&targetDependents, "target-dependents", false,
 		"Allows destroying of dependent targets discovered but not specified in --target list")
+	cmd.PersistentFlags().BoolVar(&excludeProtected, "exclude-protected", false, "Do not destroy protected resources."+
+		" Destroy all other resources.")
 
 	// Flags for engine.UpdateOptions.
 	cmd.PersistentFlags().BoolVar(
@@ -257,3 +287,52 @@ func newDestroyCmd() *cobra.Command {
 
 	return cmd
 }
+
+// seperateProtected returns a list or unprotected and protected resources respectively. This allows
+// us to safely destroy all resources in the unprotected list without invalidating any resource in
+// the protected list. Protection is contravarient: A < B where A: Protected => B: Protected, A < B
+// where B: Protected !=> A: Protected.
+//
+// A
+// B: Parent = A
+// C: Parent = A, Protect = True
+// D: Parent = C
+//
+// -->
+//
+// Unprotected: B, D
+// Protected: A, C
+//
+// We rely on the fact that `resources` is topologically sorted with respect to its dependencies.
+// This function understands that providers live outside this topological sort.
+func seperateProtected(resources []*resource.State) (
+	/*unprotected*/ []*resource.State /*protected*/, []*resource.State) {
+	dg := graph.NewDependencyGraph(resources)
+	transitiveProtected := graph.ResourceSet{}
+	for _, r := range resources {
+		if r.Protect {
+			rProtected := dg.TransitiveDependenciesOf(r)
+			rProtected[r] = true
+			transitiveProtected.UnionWith(rProtected)
+		}
+	}
+	allResources := graph.NewResourceSetFromArray(resources)
+	return allResources.SetMinus(transitiveProtected).ToArray(), transitiveProtected.ToArray()
+}
+
+// Returns the number of protected resources that remain. Appends all unprotected resources to `targetUrns`.
+func handleExcludeProtected(s backend.Stack) ([]resource.URN, int, error) {
+	// Get snapshot
+	snapshot, err := s.Snapshot(commandContext())
+	if err != nil {
+		return nil, 0, err
+	} else if snapshot == nil {
+		return nil, 0, errors.New("Failed to find the stack snapshot. Are you in a stack?")
+	}
+	unprotected, protected := seperateProtected(snapshot.Resources)
+	targetUrns := []resource.URN{}
+	for _, r := range unprotected {
+		targetUrns = append(targetUrns, r.URN)
+	}
+	return targetUrns, len(protected), nil
+}
diff --git a/pkg/resource/graph/dependency_graph.go b/pkg/resource/graph/dependency_graph.go
index 3fbd9908c..5fe908804 100644
--- a/pkg/resource/graph/dependency_graph.go
+++ b/pkg/resource/graph/dependency_graph.go
@@ -1,4 +1,4 @@
-// Copyright 2016-2018, Pulumi Corporation.  All rights reserved.
+// Copyright 2016-2021, Pulumi Corporation.  All rights reserved.
 
 package graph
 
@@ -123,6 +123,67 @@ func (dg *DependencyGraph) DependenciesOf(res *resource.State) ResourceSet {
 	return set
 }
 
+// `TransitiveDependenciesOf` calculates the set of resources that `r` depends
+// on, directly or indirectly. This includes as a `Parent`, a member of r's
+// `Dependencies` list or as a provider.
+//
+// This function is linear in the number of resources in the `DependencyGraph`.
+func (dg *DependencyGraph) TransitiveDependenciesOf(r *resource.State) ResourceSet {
+	dependentProviders := make(map[resource.URN]struct{})
+
+	urns := make(map[resource.URN]*node, len(dg.resources))
+	for _, r := range dg.resources {
+		urns[r.URN] = &node{resource: r}
+	}
+
+	// Linearity is due to short circuiting in the traversal.
+	markAsDependency(r.URN, urns, dependentProviders)
+
+	// This will only trigger if (urn, node) is a provider. The check is implicit
+	// in the set lookup.
+	for urn := range urns {
+		if _, ok := dependentProviders[urn]; ok {
+			markAsDependency(urn, urns, dependentProviders)
+		}
+	}
+
+	dependencies := ResourceSet{}
+	for _, r := range urns {
+		if r.marked {
+			dependencies[r.resource] = true
+		}
+	}
+	// We don't want to include `r` as it's own dependency.
+	delete(dependencies, r)
+	return dependencies
+
+}
+
+// Mark a resource and its parents as a dependency. This is a helper function for `TransitiveDependenciesOf`.
+func markAsDependency(urn resource.URN, urns map[resource.URN]*node, dependedProviders map[resource.URN]struct{}) {
+	r := urns[urn]
+	for {
+		r.marked = true
+		if r.resource.Provider != "" {
+			ref, err := providers.ParseReference(r.resource.Provider)
+			contract.AssertNoError(err)
+			dependedProviders[ref.URN()] = struct{}{}
+		}
+		for _, dep := range r.resource.Dependencies {
+			markAsDependency(dep, urns, dependedProviders)
+		}
+
+		// If p is already marked, we don't need to continue to traverse. All
+		// nodes above p will have already been marked. This is a property of
+		// `resources` being topologically sorted.
+		if p, ok := urns[r.resource.Parent]; ok && !p.marked {
+			r = p
+		} else {
+			break
+		}
+	}
+}
+
 // NewDependencyGraph creates a new DependencyGraph from a list of resources.
 // The resources should be in topological order with respect to their dependencies, including
 // parents appearing before children.
@@ -143,3 +204,9 @@ func NewDependencyGraph(resources []*resource.State) *DependencyGraph {
 
 	return &DependencyGraph{index, resources, childrenOf}
 }
+
+// A node in a graph.
+type node struct {
+	marked   bool
+	resource *resource.State
+}
diff --git a/pkg/resource/graph/dependency_graph_rapid_test.go b/pkg/resource/graph/dependency_graph_rapid_test.go
index 906c69339..5bcef3fe6 100644
--- a/pkg/resource/graph/dependency_graph_rapid_test.go
+++ b/pkg/resource/graph/dependency_graph_rapid_test.go
@@ -202,6 +202,24 @@ func TestRapidDependingOnOrdered(t *testing.T) {
 	}
 }
 
+func TestRapidTransitiveDependenciesOf(t *testing.T) {
+	graphCheck(t, func(t *rapid.T, universe []*resource.State) {
+		expectedInTDepsOf := transitively(universe)(expectedDependenciesOf)
+		dg := NewDependencyGraph(universe)
+		for _, a := range universe {
+			tda := dg.TransitiveDependenciesOf(a)
+			for _, b := range universe {
+				assert.Equalf(t,
+					expectedInTDepsOf(a, b),
+					tda[b],
+					"Mismatch on a=%v, b=%b",
+					a.URN,
+					b.URN)
+			}
+		}
+	})
+}
+
 // Generators --------------------------------------------------------------------------------------
 
 // Generates ordered values of type `[]ResourceState` that:
diff --git a/pkg/resource/graph/dependency_graph_test.go b/pkg/resource/graph/dependency_graph_test.go
index 73d594c72..e6710e70e 100644
--- a/pkg/resource/graph/dependency_graph_test.go
+++ b/pkg/resource/graph/dependency_graph_test.go
@@ -1,4 +1,4 @@
-// Copyright 2016-2018, Pulumi Corporation.  All rights reserved.
+// Copyright 2016-2021, Pulumi Corporation.  All rights reserved.
 
 package graph
 
@@ -229,3 +229,29 @@ func TestDependenciesOfRemoteComponentsNoCycle(t *testing.T) {
 	assert.True(t, rDependencies[parent])
 	assert.False(t, rDependencies[child])
 }
+
+func TestTransitiveDependenciesOf(t *testing.T) {
+	aws := NewProviderResource("aws", "default", "0")
+	parent := NewResource("parent", aws)
+	greatUncle := NewResource("greatUncle", aws)
+	uncle := NewResource("r", aws)
+	uncle.Parent = greatUncle.URN
+	child := NewResource("child", aws, uncle.URN)
+	child.Parent = parent.URN
+	baby := NewResource("baby", aws)
+	baby.Parent = child.URN
+
+	dg := NewDependencyGraph([]*resource.State{
+		aws,
+		parent,
+		greatUncle,
+		uncle,
+		child,
+		baby,
+	})
+	// <(relation)- as an alias for depends on via relation
+	// baby <(Parent)- child <(Dependency)- uncle <(Parent)- greatUncle <(Provider)- aws
+	set := dg.TransitiveDependenciesOf(baby)
+	assert.True(t, set[aws], "everything should depend on the provider")
+	assert.True(t, set[greatUncle], "child depends on greatUncle")
+}
diff --git a/pkg/resource/graph/resource_set.go b/pkg/resource/graph/resource_set.go
index d2e766b90..8b37234a0 100644
--- a/pkg/resource/graph/resource_set.go
+++ b/pkg/resource/graph/resource_set.go
@@ -1,4 +1,4 @@
-// Copyright 2016-2018, Pulumi Corporation.
+// Copyright 2016-2021, Pulumi Corporation.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@ package graph
 
 import (
 	"github.com/pulumi/pulumi/sdk/v3/go/common/resource"
+	"sort"
 )
 
 // ResourceSet represents a set of Resources.
@@ -32,3 +33,62 @@ func (s ResourceSet) Intersect(other ResourceSet) ResourceSet {
 
 	return newSet
 }
+
+// Returns the contents of the set as an array of resources. To ensure
+// determinism, they are sorted by urn.
+func (s ResourceSet) ToArray() []*resource.State {
+	arr := make([]*resource.State, len(s))
+	i := 0
+	for r := range s {
+		arr[i] = r
+		i++
+	}
+	sort.Slice(arr, func(i, j int) bool {
+		return arr[i].URN < arr[j].URN
+	})
+	return arr
+}
+
+// Produces a set from an array.
+func NewResourceSetFromArray(arr []*resource.State) ResourceSet {
+	s := ResourceSet{}
+	for _, r := range arr {
+		s[r] = true
+	}
+	return s
+}
+
+// Produces a shallow copy of `s`.
+func CopyResourceSet(s ResourceSet) ResourceSet {
+	result := ResourceSet{}
+	for k, v := range s {
+		result[k] = v
+	}
+	return result
+}
+
+// Computes s - other. Input sets are unchanged. If `other[k] = false`, then `k`
+// will not be removed from `s`.
+func (s ResourceSet) SetMinus(other ResourceSet) ResourceSet {
+	result := CopyResourceSet(s)
+	for k, v := range other {
+		if v {
+			delete(result, k)
+		}
+	}
+	return result
+}
+
+// Produces a new set with elements from both sets. The original sets are unchanged.
+func (s ResourceSet) Union(other ResourceSet) ResourceSet {
+	result := CopyResourceSet(s)
+	return result.UnionWith(other)
+}
+
+// Alters `s` to include elements of `other`.
+func (s ResourceSet) UnionWith(other ResourceSet) ResourceSet {
+	for k, v := range other {
+		s[k] = v || s[k]
+	}
+	return s
+}
diff --git a/pkg/resource/graph/resource_set_test.go b/pkg/resource/graph/resource_set_test.go
index 51cc4d25e..2285be65c 100644
--- a/pkg/resource/graph/resource_set_test.go
+++ b/pkg/resource/graph/resource_set_test.go
@@ -1,4 +1,4 @@
-// Copyright 2016-2018, Pulumi Corporation.
+// Copyright 2016-2021, Pulumi Corporation.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -37,3 +37,40 @@ func TestIntersect(t *testing.T) {
 	assert.True(t, setC[b])
 	assert.False(t, setC[c])
 }
+
+func TestUnion(t *testing.T) {
+	a := NewResource("a", nil)
+	b := NewResource("b", nil)
+	c := NewResource("c", nil)
+
+	setA := make(ResourceSet)
+	setA[a] = true
+	setA[c] = true
+
+	setB := make(ResourceSet)
+	setB[b] = true
+
+	setC := setA.Union(setB)
+	assert.True(t, setC[a])
+	assert.True(t, setC[b])
+	assert.True(t, setC[c])
+}
+
+func TestSetMinus(t *testing.T) {
+	a := NewResource("a", nil)
+	b := NewResource("b", nil)
+	c := NewResource("c", nil)
+
+	setA := make(ResourceSet)
+	setA[a] = true
+	setA[b] = true
+
+	setB := make(ResourceSet)
+	setB[b] = true
+	setB[c] = true
+
+	setC := setA.SetMinus(setB)
+	assert.True(t, setC[a])
+	assert.False(t, setC[b])
+	assert.False(t, setC[c])
+}
diff --git a/tests/integration/exclude_protected/Pulumi.yaml b/tests/integration/exclude_protected/Pulumi.yaml
new file mode 100644
index 000000000..13080cff7
--- /dev/null
+++ b/tests/integration/exclude_protected/Pulumi.yaml
@@ -0,0 +1,3 @@
+name: exclude-protected
+runtime: nodejs
+description: A minimal AWS TypeScript Pulumi program
diff --git a/tests/integration/exclude_protected/index.ts b/tests/integration/exclude_protected/index.ts
new file mode 100644
index 000000000..645c14894
--- /dev/null
+++ b/tests/integration/exclude_protected/index.ts
@@ -0,0 +1,35 @@
+import * as pulumi from "@pulumi/pulumi";
+
+
+class Resource extends pulumi.ComponentResource {
+    constructor(name: string, _?: {}, opts?: pulumi.ComponentResourceOptions) {
+        super("my:module:Resource", name, {}, opts);
+    }
+}
+
+
+
+const bucket1 = new Resource("my-bucket", {}, { protect: true });
+// Because `protect` is explicitly set to false, we will delete this.
+new Resource("my-bucket-child", {}, { protect: false, parent: bucket1 });
+new Resource("my-bucket-child-protected", {}, { protect: true, parent: bucket1 });
+
+const bucket2 = new Resource("my-2bucket", {}, { protect: false });
+new Resource("my-2bucket-child", {}, { protect: false, parent: bucket2 });
+new Resource("my-2bucket-protected-child", {}, { protect: true, parent: bucket2 });
+
+const p = new Resource("provided-bucket", {}, { protect: true })
+// Inherits protected status from `p`. This is protected in the state, and is thus safe.
+new Resource("provided-bucket-child", {}, { parent: p })
+new Resource("provided-bucket-child-unprotected", {}, { parent: p, protect: false })
+
+// If possible, we should do a test with providers, that looks something like
+// this. Doing a provider test with component resources is problematic because
+// `ComponentResources` don't have CRUD operations.
+//
+// import * as aws from "@pulumi/aws";
+// new aws.s3.Bucket("provider-unprotected", {}, { provider: prov })
+// const p = new aws.s3.Bucket("provided-bucket", {}, { provider: prov, protect: true })
+// // Inherits protected status from `p`. This is protected in the state, and is thus safe.
+// new aws.s3.Bucket("provided-bucket-child", {}, { parent: p })
+// new aws.s3.Bucket("provided-bucket-child-unprotected", {}, { parent: p, protect: false })
diff --git a/tests/integration/exclude_protected/package.json b/tests/integration/exclude_protected/package.json
new file mode 100644
index 000000000..ad4c90202
--- /dev/null
+++ b/tests/integration/exclude_protected/package.json
@@ -0,0 +1,9 @@
+{
+  "name": "test-protect",
+  "devDependencies": {
+    "@types/node": "^10.0.0"
+  },
+  "peerDependencies": {
+    "@pulumi/pulumi": "^3.0.0"
+  }
+}
diff --git a/tests/integration/integration_test.go b/tests/integration/integration_test.go
index db9041f5b..4172ef18b 100644
--- a/tests/integration/integration_test.go
+++ b/tests/integration/integration_test.go
@@ -943,3 +943,29 @@ func TestJSONOutputWithStreamingPreview(t *testing.T) {
 		},
 	})
 }
+
+func TestExcludeProtected(t *testing.T) {
+	e := ptesting.NewEnvironment(t)
+	defer func() {
+		if !t.Failed() {
+			e.DeleteEnvironment()
+		}
+	}()
+
+	e.ImportDirectory("exclude_protected")
+
+	e.RunCommand("pulumi", "login", "--cloud-url", e.LocalURL())
+
+	e.RunCommand("pulumi", "stack", "init", "dev")
+
+	e.RunCommand("yarn", "link", "@pulumi/pulumi")
+	e.RunCommand("yarn", "install")
+
+	e.RunCommand("pulumi", "up", "--skip-preview", "--yes")
+
+	stdout, _ := e.RunCommand("pulumi", "destroy", "--skip-preview", "--yes", "--exclude-protected")
+	assert.Contains(t, stdout, "All unprotected resources were destroyed. There are still 7 protected resources")
+	// We run the command again, but this time there are not unprotected resources to destroy.
+	stdout, _ = e.RunCommand("pulumi", "destroy", "--skip-preview", "--yes", "--exclude-protected")
+	assert.Contains(t, stdout, "There were no unprotected resources to destroy. There are still 7")
+}