diff --git a/backend/init/init.go b/backend/init/init.go
index 8c6cf0a8f..6c057c32b 100644
--- a/backend/init/init.go
+++ b/backend/init/init.go
@@ -49,7 +49,21 @@ func Backend(name string) func() backend.Backend {
 	return backends[name]
 }
 
-// NOTE(@mitchellh): At some point I'm sure we'll want Add() to add a backend
-// to the list. Right now we hardcode all backends and there isn't a reasonable
-// use case for this so we're leaving it out. It would be trivial to add in
-// the future.
+// Set sets a new backend in the list of backends. If f is nil then the
+// backend will be removed from the map. If this backend already exists
+// then it will be overwritten.
+//
+// This method sets this backend globally and care should be taken to do
+// this only before Terraform is executing to prevent odd behavior of backends
+// changing mid-execution.
+func Set(name string, f func() backend.Backend) {
+	backendsLock.Lock()
+	defer backendsLock.Unlock()
+
+	if f == nil {
+		delete(backends, name)
+		return
+	}
+
+	backends[name] = f
+}
diff --git a/builtin/providers/terraform/data_source_state.go b/builtin/providers/terraform/data_source_state.go
index 2c3c5b050..ee9a63630 100644
--- a/builtin/providers/terraform/data_source_state.go
+++ b/builtin/providers/terraform/data_source_state.go
@@ -1,11 +1,14 @@
 package terraform
 
 import (
+	"fmt"
 	"log"
 	"time"
 
+	backendinit "github.com/hashicorp/terraform/backend/init"
+	"github.com/hashicorp/terraform/config"
 	"github.com/hashicorp/terraform/helper/schema"
-	"github.com/hashicorp/terraform/state/remote"
+	"github.com/hashicorp/terraform/terraform"
 )
 
 func dataSourceRemoteState() *schema.Resource {
@@ -43,9 +46,11 @@ func dataSourceRemoteState() *schema.Resource {
 
 func dataSourceRemoteStateRead(d *schema.ResourceData, meta interface{}) error {
 	backend := d.Get("backend").(string)
-	config := make(map[string]string)
-	for k, v := range d.Get("config").(map[string]interface{}) {
-		config[k] = v.(string)
+
+	// Get the configuration in a type we want.
+	rawConfig, err := config.NewRawConfig(d.Get("config").(map[string]interface{}))
+	if err != nil {
+		return fmt.Errorf("error initializing backend: %s", err)
 	}
 
 	// Don't break people using the old _local syntax - but note warning above
@@ -55,15 +60,23 @@ func dataSourceRemoteStateRead(d *schema.ResourceData, meta interface{}) error {
 	}
 
 	// Create the client to access our remote state
-	log.Printf("[DEBUG] Initializing remote state client: %s", backend)
-	client, err := remote.NewClient(backend, config)
-	if err != nil {
-		return err
+	log.Printf("[DEBUG] Initializing remote state backend: %s", backend)
+	f := backendinit.Backend(backend)
+	if f == nil {
+		return fmt.Errorf("Unknown backend type: %s", backend)
+	}
+	b := f()
+
+	// Configure the backend
+	if err := b.Configure(terraform.NewResourceConfig(rawConfig)); err != nil {
+		return fmt.Errorf("error initializing backend: %s", err)
 	}
 
-	// Create the remote state itself and refresh it in order to load the state
-	log.Printf("[DEBUG] Loading remote state...")
-	state := &remote.State{Client: client}
+	// Get the state
+	state, err := b.State()
+	if err != nil {
+		return fmt.Errorf("error loading the remote state: %s", err)
+	}
 	if err := state.RefreshState(); err != nil {
 		return err
 	}
diff --git a/builtin/providers/terraform/data_source_state_test.go b/builtin/providers/terraform/data_source_state_test.go
index 1d56823ef..9a58b154f 100644
--- a/builtin/providers/terraform/data_source_state_test.go
+++ b/builtin/providers/terraform/data_source_state_test.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"testing"
 
+	backendinit "github.com/hashicorp/terraform/backend/init"
 	"github.com/hashicorp/terraform/helper/resource"
 	"github.com/hashicorp/terraform/terraform"
 )
@@ -24,6 +25,25 @@ func TestState_basic(t *testing.T) {
 	})
 }
 
+func TestState_backends(t *testing.T) {
+	backendinit.Set("_ds_test", backendinit.Backend("local"))
+	defer backendinit.Set("_ds_test", nil)
+
+	resource.UnitTest(t, resource.TestCase{
+		PreCheck:  func() { testAccPreCheck(t) },
+		Providers: testAccProviders,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccState_backend,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStateValue(
+						"data.terraform_remote_state.foo", "foo", "bar"),
+				),
+			},
+		},
+	})
+}
+
 func TestState_complexOutputs(t *testing.T) {
 	resource.UnitTest(t, resource.TestCase{
 		PreCheck:  func() { testAccPreCheck(t) },
@@ -72,6 +92,15 @@ data "terraform_remote_state" "foo" {
 	}
 }`
 
+const testAccState_backend = `
+data "terraform_remote_state" "foo" {
+	backend = "_ds_test"
+
+	config {
+		path = "./test-fixtures/basic.tfstate"
+	}
+}`
+
 const testAccState_complexOutputs = `
 resource "terraform_remote_state" "foo" {
 	backend = "local"