diff --git a/plans/backup.pp b/plans/backup.pp
index 977457be..6672f3dd 100644
--- a/plans/backup.pp
+++ b/plans/backup.pp
@@ -1,3 +1,4 @@
+# @api private
 # @summary Backup the core user settings for puppet infrastructure
 #
 # This plan can backup data as outlined at insert doc
@@ -110,7 +111,8 @@
 
   run_command(@("CMD"/L), $primary_target)
     umask 0077 \
-      && tar -czf ${shellquote($backup_directory)}.tar.gz ${shellquote($backup_directory)} \
+      && cd ${shellquote(dirname($backup_directory))} \
+      && tar -czf ${shellquote($backup_directory)}.tar.gz ${shellquote(basename($backup_directory))} \
       && rm -rf ${shellquote($backup_directory)}
     | CMD
 
diff --git a/plans/restore.pp b/plans/restore.pp
new file mode 100644
index 00000000..68a7cd9b
--- /dev/null
+++ b/plans/restore.pp
@@ -0,0 +1,235 @@
+# @api private
+# @summary Restore the core user settings for puppet infrastructure from backup
+#
+# This plan can restore data to puppet infrastructure for DR and rebuilds
+# 
+plan peadm::restore (
+  # This plan should be run on the primary server
+  Peadm::SingleTargetSpec $targets,
+
+  # Which data to restore
+  Peadm::Recovery_opts    $restore = {},
+
+  # Path to the recovery tarball
+  Pattern[/.*\.tar\.gz$/] $input_file,
+) {
+  peadm::assert_supported_bolt_version()
+
+  $recovery_opts = (peadm::recovery_opts_default() + $restore)
+  $cluster = run_task('peadm::get_peadm_config', $targets).first.value
+  $arch = peadm::assert_supported_architecture(
+    getvar('cluster.params.primary_host'),
+    getvar('cluster.params.replica_host'),
+    getvar('cluster.params.primary_postgresql_host'),
+    getvar('cluster.params.replica_postgresql_host'),
+    getvar('cluster.params.compiler_hosts'),
+  )
+
+  $primary_target   = peadm::get_targets(getvar('cluster.params.primary_host'), 1)
+  $replica_target   = peadm::get_targets(getvar('cluster.params.replica_host'), 1)
+  $compiler_targets = peadm::get_targets(getvar('cluster.params.compiler_hosts'))
+
+  # Determine the array of targets to which the PuppetDB PostgreSQL database
+  # should be restored to. This could be as simple as just the primary server,
+  # or it could be two separate PostgreSQL servers.
+  $puppetdb_postgresql_targets = peadm::flatten_compact([
+    getvar('cluster.params.primary_postgresql_host') ? {
+      undef   => $primary_target,
+      default => peadm::get_targets(getvar('cluster.params.primary_postgresql_host'), 1),
+    },
+    getvar('cluster.params.replica_postgresql_host') ? {
+      undef   => $replica_target,
+      default => peadm::get_targets(getvar('cluster.params.replica_postgresql_host'), 1),
+    },
+  ])
+
+  $puppetdb_targets = peadm::flatten_compact([
+    $primary_target,
+    $replica_target,
+    $compiler_targets,
+  ])
+
+  $recovery_directory = "${dirname($input_file)}/${basename("${input_file}", '.tar.gz')}"
+
+  run_command(@("CMD"/L), $primary_target)
+    umask 0077 \
+      && cd ${shellquote(dirname($recovery_directory))} \
+      && tar -xzf ${shellquote($input_file)}
+    | CMD
+
+  # Map of recovery option name to array of database hosts to restore the
+  # relevant .dump content to.
+  $restore_databases = {
+    'orchestrator' => [$primary_target],
+    'activity'     => [$primary_target],
+    'rbac'         => [$primary_target],
+    'puppetdb'     => $puppetdb_postgresql_targets,
+  }.filter |$key,$_| {
+    $recovery_opts[$key] == true
+  }
+
+  if getvar('recovery_opts.classifier') {
+    out::message('# Restoring classification')
+    run_task('peadm::backup_classification', $primary_target,
+      directory => $recovery_directory
+    )
+    out::message("# Backed up current classification to ${recovery_directory}/classification_backup.json")
+
+    run_task('peadm::transform_classification_groups', $primary_target,
+      source_directory  => "${recovery_directory}/classifier",
+      working_directory => $recovery_directory
+    )
+
+    run_task('peadm::restore_classification', $primary_target,
+      classification_file => "${recovery_directory}/classification_backup.json",
+    )
+  }
+
+  if getvar('recovery_opts.ca') {
+    out::message('# Restoring ca and ssl certificates')
+    run_command(@("CMD"/L), $primary_target)
+      /opt/puppetlabs/bin/puppet-backup restore \
+        --scope=certs \
+        --tempdir=${shellquote($recovery_directory)} \
+        --force \
+        ${shellquote($recovery_directory)}/classifier/pe_backup-*tgz
+      | CMD
+  }
+
+  # Use PuppetDB's /pdb/admin/v1/archive API to SAVE data currently in PuppetDB.
+  # Otherwise we'll completely lose it if/when we restore.
+  # TODO: consider adding a heuristic to skip when innappropriate due to size
+  #       or other factors.
+  if getvar('recovery_opts.puppetdb') {
+    run_command(@("CMD"/L), $primary_target)
+      /opt/puppetlabs/bin/puppet-db export ${shellquote($recovery_directory)}/puppetdb-archive.bin
+      | CMD
+  }
+
+  ## shutdown services
+  run_command(@("CMD"/L), $primary_target)
+    systemctl stop pe-console-services pe-nginx pxp-agent pe-puppetserver \
+                   pe-orchestration-services puppet pe-puppetdb
+    | CMD
+
+  # Restore secrets/keys.json if it exists
+  out::message('# Restoring ldap secret key if it exists')
+  run_command(@("CMD"/L), $primary_target)
+    test -f ${shellquote($recovery_directory)}/rbac/keys.json \
+      && cp -rp ${shellquote($recovery_directory)}/keys.json /etc/puppetlabs/console-services/conf.d/secrets/ \
+      || echo secret ldap key doesnt exist
+    | CMD
+
+  # IF restoring orchestrator restore the secrets to /etc/puppetlabs/orchestration-services/conf.d/secrets/
+  if getvar('recovery_opts.orchestrator') {
+    out::message('# Restoring orchestrator secret keys')
+    run_command(@("CMD"/L), $primary_target)
+      cp -rp ${shellquote($recovery_directory)}/orchestrator/secrets/* /etc/puppetlabs/orchestration-services/conf.d/secrets/
+      | CMD
+  }
+
+  #$database_to_restore.each |Integer $index, Boolean $value | {
+  $restore_databases.each |$name,$database_targets| {
+    out::message("# Restoring ${name} database")
+    $dbname = "pe-${shellquote($name)}"
+
+    # Drop pglogical extensions and schema if present
+    run_command(@("CMD"/L), $database_targets)
+      su - pe-postgres -s /bin/bash -c \
+        "/opt/puppetlabs/server/bin/psql \
+           --tuples-only \
+           -d '${dbname}' \
+           -c 'DROP SCHEMA IF EXISTS pglogical CASCADE;'"
+      | CMD
+
+    run_command(@("CMD"/L), $database_targets)
+      su - pe-postgres -s /bin/bash -c \
+        "/opt/puppetlabs/server/bin/psql \
+           -d '${dbname}' \
+           -c 'DROP SCHEMA public CASCADE; CREATE SCHEMA public;'"
+      | CMD
+
+    # To allow db user to restore the database grant temporary privileges
+    run_command(@("CMD"/L), $database_targets)
+      su - pe-postgres -s /bin/bash -c \
+        "/opt/puppetlabs/server/bin/psql \
+           -d '${dbname}' \
+           -c 'ALTER USER \"${dbname}\" WITH SUPERUSER;'"
+      | CMD
+
+    # Restore database. If there are multiple database restore targets, perform
+    # the restore(s) in parallel.
+    parallelize($database_targets) |$database_target| {
+      run_command(@("CMD"/L), $primary_target)
+        /opt/puppetlabs/server/bin/pg_restore \
+          -j 4 \
+          -d "sslmode=verify-ca \
+              host=${shellquote($database_target.peadm::certname())} \
+              sslcert=/etc/puppetlabs/puppetdb/ssl/${shellquote($primary_target.peadm::certname())}.cert.pem \
+              sslkey=/etc/puppetlabs/puppetdb/ssl/${shellquote($primary_target.peadm::certname())}.private_key.pem \
+              sslrootcert=/etc/puppetlabs/puppet/ssl/certs/ca.pem \
+              dbname=${dbname} \
+              user=${dbname}" \
+          -Fd ${recovery_directory}/${name}/${dbname}.dump.d
+        | CMD
+    }
+
+    # Remove db user privileges post restore
+    run_command(@("CMD"/L), $database_targets)
+      su - pe-postgres -s /bin/bash -c \
+        "/opt/puppetlabs/server/bin/psql \
+           -d '${dbname}' \
+           -c 'ALTER USER \"${dbname}\" WITH NOSUPERUSER;'"
+      | CMD
+
+    # Drop pglogical extension and schema (again) if present after db restore
+    run_command(@("CMD"/L), $database_targets)
+      su - pe-postgres -s /bin/bash -c \
+        "/opt/puppetlabs/server/bin/psql \
+           --tuples-only \
+           -d '${dbname}' \
+           -c 'DROP SCHEMA IF EXISTS pglogical CASCADE;'"
+      | CMD
+
+    run_command(@("CMD"/L), $database_targets)
+      su - pe-postgres -s /bin/bash -c \
+        "/opt/puppetlabs/server/bin/psql \
+           -d '${dbname}' \
+           -c 'DROP EXTENSION IF EXISTS pglogical CASCADE;'"
+      | CMD
+  }
+
+  # Use `puppet infra` to ensure correct file permissions, restart services,
+  # etc. Make sure not to try and get config data from the classifier, which
+  # isn't yet up and running.
+  run_command(@("CMD"/L), $primary_target)
+    /opt/puppetlabs/bin/puppet-infrastructure configure --no-recover
+    | CMD
+
+  # If we have replicas reinitalise them
+  run_command(@("CMD"/L), $replica_target)
+    /opt/puppetlabs/bin/puppet-infra reinitialize replica -y
+    | CMD
+
+  # Use PuppetDB's /pdb/admin/v1/archive API to MERGE previously saved data
+  # into the restored database.
+  # TODO: consider adding a heuristic to skip when innappropriate due to size
+  #       or other factors.
+  if getvar('recovery_opts.puppetdb') {
+    run_command(@("CMD"/L), $primary_target)
+      /opt/puppetlabs/bin/puppet-db import ${shellquote($recovery_directory)}/puppetdb-archive.bin
+      | CMD
+  }
+
+  # Run Puppet to pick up last remaining config tweaks
+  run_task('peadm::puppet_runonce', $primary_target)
+
+  apply($primary_target){
+    file { $recovery_directory :
+      ensure => 'absent',
+      force  => true
+    }
+  }
+
+  return("success")
+}
diff --git a/spec/plans/restore_spec.rb b/spec/plans/restore_spec.rb
new file mode 100644
index 00000000..47763365
--- /dev/null
+++ b/spec/plans/restore_spec.rb
@@ -0,0 +1,24 @@
+require 'spec_helper'
+
+describe 'peadm::restore' do
+  include BoltSpec::Plans
+  let(:params) { { 'primary_host' => 'primary', 'backup_timestamp' => '2022-03-29_16:57:41' } }
+
+  it 'runs with default params' do
+    allow_apply
+    pending('a lack of support for functions requires a workaround to be written')
+    expect_task('peadm::get_peadm_config').always_return({ 'primary_postgresql_host' => 'postgres' })
+    expect_out_message.with_params('# Backing up ca and ssl certificates')
+    # The commands all have a timestamp in them and frankly its proved to hard with bolt spec to work this out
+    allow_any_command
+    expect_out_message.with_params('# Restoring classification')
+    expect_out_message.with_params('# Backed up current classification to /tmp/classification_backup.json')
+    expect_out_message.with_params('# Restoring ca and ssl certificates')
+    expect_out_message.with_params('# Restoring database pe-orchestrator')
+    expect_out_message.with_params('# Restoring database pe-activity')
+    expect_out_message.with_params('# Restoring database pe-rbac')
+    expect_out_message.with_params('# Restoring classification')
+    expect_task('peadm::backup_classification')
+    expect(run_plan('peadm::restore', params)).to be_ok
+  end
+end
diff --git a/tasks/get_peadm_config.rb b/tasks/get_peadm_config.rb
index e215649c..52f78057 100755
--- a/tasks/get_peadm_config.rb
+++ b/tasks/get_peadm_config.rb
@@ -18,13 +18,19 @@ def config
     # Compute values
     primary = groups.pinned('PE Master')
     replica = groups.pinned('PE HA Replica')
-    server_a = server('puppet/server', 'A')
-    server_b = server('puppet/server', 'B')
+    server_a = server('puppet/server', 'A', [primary, replica].compact)
+    server_b = server('puppet/server', 'B', [primary, replica].compact)
     primary_letter = primary.eql?(server_a) ? 'A' : 'B'
     replica_letter = primary_letter.eql?('A') ? 'B' : 'A'
+
+    configured_postgresql_servers = [
+      groups.dig('PE Primary A', 'config_data', 'puppet_enterprise::profile::puppetdb', 'database_host'),
+      groups.dig('PE Primary B', 'config_data', 'puppet_enterprise::profile::puppetdb', 'database_host'),
+    ].compact
+
     postgresql = {
-      'A' => server('puppet/puppetdb-database', 'A'),
-      'B' => server('puppet/puppetdb-database', 'B'),
+      'A' => server('puppet/puppetdb-database', 'A', configured_postgresql_servers),
+      'B' => server('puppet/puppetdb-database', 'B', configured_postgresql_servers),
     }
 
     # Build and return the task output
@@ -78,10 +84,11 @@ def compilers
     end
   end
 
-  def server(role, letter)
+  def server(role, letter, certname_array)
     query = 'inventory[certname] { '\
             '  trusted.extensions."1.3.6.1.4.1.34380.1.1.9812" = "' + role + '" and ' \
-            '  trusted.extensions."1.3.6.1.4.1.34380.1.1.9813" = "' + letter + '"}'
+            '  trusted.extensions."1.3.6.1.4.1.34380.1.1.9813" = "' + letter + '" and ' \
+            '  certname in ' + certname_array.to_json + '}'
 
     server = pdb_query(query).map { |n| n['certname'] }
     raise "More than one #{letter} #{role} server found!" unless server.size <= 1
diff --git a/tasks/restore_classification.json b/tasks/restore_classification.json
new file mode 100644
index 00000000..3f527ce3
--- /dev/null
+++ b/tasks/restore_classification.json
@@ -0,0 +1,12 @@
+{
+  "puppet_task_version": 1,
+  "supports_noop": false,
+  "description": "A short description of this task",
+  "parameters": {
+    "classification_file": {
+      "type": "String",
+      "description": "The full path to a backed up or transformed classification file",
+      "default": "/tmp/transformed_classification.json"
+    }
+  }
+}
diff --git a/tasks/restore_classification.rb b/tasks/restore_classification.rb
new file mode 100755
index 00000000..cf08a248
--- /dev/null
+++ b/tasks/restore_classification.rb
@@ -0,0 +1,45 @@
+#!/opt/puppetlabs/puppet/bin/ruby
+
+# Puppet Task Name: backup_classification
+require 'net/https'
+require 'uri'
+require 'json'
+require 'puppet'
+
+# RestoreClassifiation task class
+class RestoreClassification
+  def initialize(params)
+    @classification_file = params['classification_file']
+  end
+
+  def execute!
+    restore_classification
+    puts "Classification restored from #{@classification_file}"
+  end
+
+  private
+
+  def https_client
+    client = Net::HTTP.new('localhost', '4433')
+    client.use_ssl = true
+    client.cert = @cert ||= OpenSSL::X509::Certificate.new(File.read(Puppet.settings[:hostcert]))
+    client.key = @key ||= OpenSSL::PKey::RSA.new(File.read(Puppet.settings[:hostprivkey]))
+    client.verify_mode = OpenSSL::SSL::VERIFY_NONE
+    client
+  end
+
+  def restore_classification
+    classification = https_client
+    classification_post = Net::HTTP::Post.new('/classifier-api/v1/import-hierarchy', 'Content-Type' => 'application/json')
+    classification_post.body = File.read(@classification_file)
+    classification.request(classification_post)
+  end
+end
+# Run the task unless an environment flag has been set, signaling not to. The
+# environment flag is used to disable auto-execution and enable Ruby unit
+# testing of this task.
+unless ENV['RSPEC_UNIT_TEST_MODE']
+  Puppet.initialize_settings
+  task = RestoreClassification.new(JSON.parse(STDIN.read))
+  task.execute!
+end
diff --git a/tasks/transform_classification_groups.json b/tasks/transform_classification_groups.json
new file mode 100644
index 00000000..4284d04f
--- /dev/null
+++ b/tasks/transform_classification_groups.json
@@ -0,0 +1,17 @@
+{
+    "description": "Transform the user groups from a source backup to a list of groups on the target server",
+    "parameters": {
+      "source_directory": {
+        "type": "String",
+        "description": "Location of Source node group yaml file"
+      },
+      "working_directory": {
+        "type": "String",
+        "description": "Location of target node group yaml file and where to create the transformed file"
+      }
+    },
+    "input_method": "stdin",
+    "implementations": [
+      {"name": "transform_classification_groups.py"}
+    ]
+  }
\ No newline at end of file
diff --git a/tasks/transform_classification_groups.py b/tasks/transform_classification_groups.py
new file mode 100644
index 00000000..ca5355b7
--- /dev/null
+++ b/tasks/transform_classification_groups.py
@@ -0,0 +1,79 @@
+#!/usr/bin/env python
+"""This module takes two classification outputs from source and targer puppet infrastructure and
+takes the user defintions from the source and adds them to the infrastructure defintions of the
+target. Allowing the ability to restore a backup of user node definitions"""
+
+import json
+import sys
+params = json.load(sys.stdin)
+source_classification_file      = params['source_directory']+"/classification_backup.json"
+target_classification_file      = params['working_directory']+"/classification_backup.json"
+transformed_classification_file = params['working_directory']+"/transformed_classification.json"
+
+def removesubgroups(data_rsg,id_rsg):
+    """
+    This definition allows us to traverse recursively down the json groups finding all children of
+    the pe infrastructure and to remove them.
+
+    Inputs are the resource group and parent ID of the resource groups
+
+    Returns
+    -------
+    data_rsg : list
+       The resource groups which did not have the parent ID
+    """
+    groups = list(filter(lambda x:x ["parent"]==id_rsg,data_rsg))
+    for group in groups:
+        subid = group["id"]
+        data_rsg = list(filter(lambda x:x ["id"]!=subid,data_rsg)) # pylint: disable=cell-var-from-loop
+        data_rsg = removesubgroups(data_rsg,subid)
+    return data_rsg
+
+# This defintion allows us to traverse down the pe inf tree and find all groups
+def addsubgroups(data_asg,id_asg,peinf_asg):
+    """
+    This definition allows us to traverse recursively down the json groups finding all groups in
+    the pe infrastructure tree and adding them to a list recursively and then returning the list.
+
+    Inputs are the list of all resource groups, infrastructure resource groups found so far and
+    parent ID of infrastructure groups
+
+    Returns
+    -------
+    data_asg : list
+       The list of resource groups of pe infrastructure groups at source
+    """
+    groups = list(filter(lambda x:x ["parent"]==id_asg,data_asg))
+    peinf_asg = peinf_asg + groups
+    for group in groups:
+        subid = group["id"]
+        peinf_asg = addsubgroups(data_asg,subid,peinf_asg)
+    return peinf_asg
+
+# open the backup classification
+with open(source_classification_file) as data_file:
+    data = json.load(data_file)
+# open the DR server classification
+with open(target_classification_file) as data_fileDR:
+    data_DR = json.load(data_fileDR)
+
+
+# find the infrastructure group and its ID
+peinf = list(filter(lambda x:x ["name"]=="PE Infrastructure",data))
+group_id = peinf[0]["id"]
+# remove this group from the list and recursively remove all sub groups
+data = list(filter(lambda x:x ["id"]!=group_id,data))
+data = removesubgroups(data,group_id)
+
+# find the dr infrastructure group and its ID
+peinf_DR = list(filter(lambda x:x ["name"]=="PE Infrastructure",data_DR))
+id_DR = peinf_DR[0]["id"]
+# Recursively go through inf groups to get the full tree
+peinf_DR = addsubgroups(data_DR,id_DR,peinf_DR)
+
+# Add the contents of the backup classification without pe inf to the DR pe inf groups
+# and write to a file
+peinf_transformed_groups = data + peinf_DR
+with open(transformed_classification_file, 'w') as fp:
+    json.dump(peinf_transformed_groups, fp)
+    
\ No newline at end of file