Repository: google/exposure-notifications-server
Pull request #1044 (2020-10-01 04:25:16)

Allow for importing of export files from other servers.
Towards #928 

## Proposed Changes

* To aid in migration from one key server infrastructure to another
* Verifies signatures
* Handles insert and revise

TODO: terraform to set this up for invocation

**Release Note**

```release-note
Add new server that can import export files generated from other key servers (including possibly other implementations).
```

google-oss-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/google/exposure-notifications-server/pull/1044#" title="Author self-approved">mikehelmick</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=google%2Fexposure-notifications-server).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/google/exposure-notifications-server/blob/main/OWNERS)~~ [mikehelmick]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
mikehelmick: I still have to run through some test cases, but this is working.
google-oss-robot: @mikehelmick: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-en-server-release-unit | 8362a925ab1a1e5bd19e264c961f85f78d6c0328 | [link](https://oss-prow.knative.dev/view/gs/oss-prow/pr-logs/pull/google_exposure-notifications-server/1044/pull-en-server-release-unit/1311522704687370240) | `/test pull-en-server-release-unit`

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
google-oss-robot: @mikehelmick: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-en-server-release-unit | eebcd83afeb2bcf0a9553ccf483865e6fbde86c5 | [link](https://oss-prow.knative.dev/view/gs/oss-prow/pr-logs/pull/google_exposure-notifications-server/1044/pull-en-server-release-unit/1311523963049873408) | `/test pull-en-server-release-unit`

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/cmd/export-importer/main.go b/cmd/export-importer/main.go
new file mode 100644
index 0000000..7e59c8d
--- /dev/null
+++ b/cmd/export-importer/main.go
@@ -0,0 +1,72 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// This binary provides a server that can import export files from a trusted partner.
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+
+	"github.com/google/exposure-notifications-server/internal/buildinfo"
+	"github.com/google/exposure-notifications-server/internal/exportimport"
+	"github.com/google/exposure-notifications-server/internal/setup"
+	"github.com/google/exposure-notifications-server/pkg/logging"
+	"github.com/google/exposure-notifications-server/pkg/server"
+	"github.com/sethvargo/go-signalcontext"
+)
+
+func main() {
+	ctx, done := signalcontext.OnInterrupt()
+
+	debug, _ := strconv.ParseBool(os.Getenv("LOG_DEBUG"))
+	logger := logging.NewLogger(debug)
+	logger = logger.With("build_id", buildinfo.BuildID)
+	logger = logger.With("build_tag", buildinfo.BuildTag)
+
+	ctx = logging.WithLogger(ctx, logger)
+
+	err := realMain(ctx)
+	done()
+
+	if err != nil {
+		logger.Fatal(err)
+	}
+}
+
+func realMain(ctx context.Context) error {
+	logger := logging.FromContext(ctx)
+
+	var config exportimport.Config
+	env, err := setup.Setup(ctx, &config)
+	if err != nil {
+		return fmt.Errorf("setup.Setup: %w", err)
+	}
+	defer env.Close(ctx)
+
+	rotationServer, err := exportimport.NewServer(&config, env)
+	if err != nil {
+		return fmt.Errorf("exportimport.NewServer: %w", err)
+	}
+
+	srv, err := server.New(config.Port)
+	if err != nil {
+		return fmt.Errorf("server.New: %w", err)
+	}
+	logger.Info("listening on: ", config.Port)
+
+	return srv.ServeHTTPHandler(ctx, rotationServer.Routes(ctx))
+}
diff --git a/internal/export/exportfile.go b/internal/export/exportfile.go
index ecfffc5..6e13b95 100644
--- a/internal/export/exportfile.go
+++ b/internal/export/exportfile.go
@@ -91,10 +91,14 @@ func MarshalExportFile(eb *model.ExportBatch, exposures, revisedExposures []*pub
 }
 
 // UnmarshalExportFile extracts the protobuf encoded exposure key present in the zip archived payload.
-func UnmarshalExportFile(zippedProtoPayload []byte) (*export.TemporaryExposureKeyExport, error) {
+// Returns the parsed TemporaryExposureKeyExport protocol buffer message, the SHA256 digest of the signed content
+// and/or an error if error.
+// The digest is useful in validating the signature as it returns the deigest of the content that
+// was signed when the archive was created.
+func UnmarshalExportFile(zippedProtoPayload []byte) (*export.TemporaryExposureKeyExport, []byte, error) {
 	zp, err := zip.NewReader(bytes.NewReader(zippedProtoPayload), int64(len(zippedProtoPayload)))
 	if err != nil {
-		return nil, fmt.Errorf("can't read payload: %v", err)
+		return nil, nil, fmt.Errorf("can't read payload: %v", err)
 	}
 
 	for _, file := range zp.File {
@@ -103,33 +107,35 @@ func UnmarshalExportFile(zippedProtoPayload []byte) (*export.TemporaryExposureKe
 		}
 	}
 
-	return nil, fmt.Errorf("payload is invalid: no %v file was found", exportBinaryName)
+	return nil, nil, fmt.Errorf("payload is invalid: no %v file was found", exportBinaryName)
 }
 
-func unmarshalContent(file *zip.File) (*export.TemporaryExposureKeyExport, error) {
+func unmarshalContent(file *zip.File) (*export.TemporaryExposureKeyExport, []byte, error) {
 	f, err := file.Open()
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	defer f.Close()
 
 	content, err := ioutil.ReadAll(f)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
+	digest := sha256.Sum256(content)
+
 	prefix := content[:fixedHeaderWidth]
 	if !bytes.Equal(prefix, fixedHeader) {
-		return nil, fmt.Errorf("unknown prefix: %v", string(prefix))
+		return nil, nil, fmt.Errorf("unknown prefix: %v", string(prefix))
 	}
 
 	message := new(export.TemporaryExposureKeyExport)
 	err = proto.Unmarshal(content[fixedHeaderWidth:], message)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	return message, nil
+	return message, digest[:], nil
 }
 
 func sortExposures(exposures []*publishmodel.Exposure) {
diff --git a/internal/exportimport/config.go b/internal/exportimport/config.go
new file mode 100644
index 0000000..fae5c10
--- /dev/null
+++ b/internal/exportimport/config.go
@@ -0,0 +1,59 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exportimport
+
+import (
+	"time"
+
+	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/setup"
+	"github.com/google/exposure-notifications-server/pkg/observability"
+)
+
+var _ setup.DatabaseConfigProvider = (*Config)(nil)
+var _ setup.ObservabilityExporterConfigProvider = (*Config)(nil)
+
+type Config struct {
+	Database              database.Config
+	ObservabilityExporter observability.Config
+
+	Port string `env:"PORT, default=8080"`
+
+	IndexFileDownloadTimeout  time.Duration `env:"INDEX_FILE_DOWNLOAD_TIMEOUT, default=30s"`
+	ExportFileDownloadTimeout time.Duration `env:"EXPORT_FILE_DOWNLOAD_TIMEOUT, default=2m"`
+
+	MaxInsertBatchSize           int           `env:"MAX_INSERT_BATCH_SIZE, default=100"`
+	MaxIntervalAge               time.Duration `env:"MAX_INTERVAL_AGE_ON_PUBLISH, default=360h"`
+	MaxMagnitudeSymptomOnsetDays uint          `env:"MAX_SYMPTOM_ONSET_DAYS, default=14"`
+	CreatedAtTruncateWindow      time.Duration `env:"TRUNCATE_WINDOW, default=1h"`
+	// Each import config is locked while file are being imported. This is to prevent starvation in
+	// the event that there are multiple exportimport configurations. If a worker fails
+	// during processing, this defines the lock timeout.
+	ImportLockTime time.Duration `env:"IMPORT_LOCK_TIME, default=13m"`
+	// Maximum amount of time that an import worker should be allowed to run. This should be set
+	// lower that infrastructure level request timeouts and lower than the lock time.
+	MaxRuntime time.Duration `env:"MAX_RUNTIME, default=12m"`
+	// Each exposure is inserted with the app_package_name / healthAuthorityID that it was published with
+	// Use this string to signal that a key came from the export-importer job.
+	ImportAPKName string `env:"IMPORT_APP_PACKAGE_NAME, default=exportimport"`
+}
+
+func (c *Config) DatabaseConfig() *database.Config {
+	return &c.Database
+}
+
+func (c *Config) ObservabilityExporterConfig() *observability.Config {
+	return &c.ObservabilityExporter
+}
diff --git a/internal/exportimport/database/exportimport.go b/internal/exportimport/database/exportimport.go
new file mode 100644
index 0000000..3238565
--- /dev/null
+++ b/internal/exportimport/database/exportimport.go
@@ -0,0 +1,423 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package database is a database interface for export importing.
+package database
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/exportimport/model"
+	"github.com/jackc/pgx/v4"
+)
+
+type ExportImportDB struct {
+	db *database.DB
+}
+
+func New(db *database.DB) *ExportImportDB {
+	return &ExportImportDB{
+		db: db,
+	}
+}
+
+func (db *ExportImportDB) ActiveConfigs(ctx context.Context) ([]*model.ExportImport, error) {
+	var configs []*model.ExportImport
+
+	if err := db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		rows, err := tx.Query(ctx, `
+			SELECT
+				id, index_file, export_root, region, from_timestamp, thru_timestamp
+			FROM
+				exportimport
+			WHERE
+				from_timestamp <= $1
+			AND
+				(thru_timestamp IS NULL OR thru_timestamp >= $1)
+		`, time.Now().UTC())
+		if err != nil {
+			return fmt.Errorf("failed to list: %w", err)
+		}
+		defer rows.Close()
+
+		for rows.Next() {
+			if err := rows.Err(); err != nil {
+				return fmt.Errorf("failed to iterate: %w", err)
+			}
+
+			cfg, err := scanOneConfig(rows)
+			if err != nil {
+				return err
+			}
+			configs = append(configs, cfg)
+		}
+		return nil
+	}); err != nil {
+		return nil, fmt.Errorf("listing active exportimport configs: %w", err)
+	}
+
+	return configs, nil
+}
+
+func scanOneConfig(row pgx.Row) (*model.ExportImport, error) {
+	var (
+		m    model.ExportImport
+		thru *time.Time
+	)
+
+	if err := row.Scan(&m.ID, &m.IndexFile, &m.ExportRoot, &m.Region, &m.From, &thru); err != nil {
+		return nil, err
+	}
+	if thru != nil {
+		m.Thru = thru
+	}
+
+	return &m, nil
+}
+
+func (db *ExportImportDB) AddConfig(ctx context.Context, ei *model.ExportImport) error {
+	if err := ei.Validate(); err != nil {
+		return err
+	}
+
+	return db.db.InTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		row := tx.QueryRow(ctx, `
+			INSERT INTO
+			ExportImport
+				(index_file, export_root, region, from_timestamp, thru_timestamp)
+			VALUES
+				($1, $2, $3, $4, $5)
+			RETURNING id
+		`, ei.IndexFile, ei.ExportRoot, ei.Region, ei.From, ei.Thru)
+
+		if err := row.Scan(&ei.ID); err != nil {
+			return fmt.Errorf("fetching exportimport.ID: %w", err)
+		}
+		return nil
+	})
+}
+
+func (db *ExportImportDB) ExpireImportFilePublicKey(ctx context.Context, ifpk *model.ImportFilePublicKey) error {
+	return db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		now := time.Now().UTC()
+		if thru := ifpk.Thru; thru != nil && now.After(*thru) {
+			return fmt.Errorf("importfilepublic key is already expired")
+		}
+		ifpk.Thru = &now
+		result, err := tx.Exec(ctx, `
+			UPDATE ImportFilePublicKey
+			SET
+				thru_timestamp = $1
+			WHERE
+				export_import_id = $2 AND
+				key_id = $3 AND
+				key_version = $4
+			`, ifpk.Thru, ifpk.ExportImportID, ifpk.KeyID, ifpk.KeyVersion)
+		if err != nil {
+			return fmt.Errorf("expiring importfilepublickey: %w", err)
+		}
+		if result.RowsAffected() != 1 {
+			return fmt.Errorf("importfilepublickey not found while expiring")
+		}
+		return nil
+	})
+}
+
+func (db *ExportImportDB) AddImportFilePublicKey(ctx context.Context, ifpk *model.ImportFilePublicKey) error {
+	return db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		result, err := tx.Exec(ctx, `
+			INSERT INTO
+				ImportFilePublicKey
+				(export_import_id, key_id, key_version, public_key, from_timestamp, thru_timestamp)
+			VALUES
+				($1, $2, $3, $4, $5, $6)
+			`, ifpk.ExportImportID, ifpk.KeyID, ifpk.KeyVersion, ifpk.PublicKeyPEM, ifpk.From, ifpk.Thru)
+
+		if err != nil {
+			return fmt.Errorf("inserting importfilepublickey: %w", err)
+		}
+		if result.RowsAffected() != 1 {
+			return fmt.Errorf("no rows inserted")
+		}
+		return nil
+	})
+}
+
+func (db *ExportImportDB) AllowedKeys(ctx context.Context, ei *model.ExportImport) ([]*model.ImportFilePublicKey, error) {
+	var publicKeys []*model.ImportFilePublicKey
+
+	if err := db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		rows, err := tx.Query(ctx, `
+			SELECT
+				export_import_id, key_id, key_version, public_key, from_timestamp, thru_timestamp
+			FROM
+				ImportFilePublicKey
+			WHERE
+				export_import_id = $1
+			AND
+				(thru_timestamp IS NULL OR thru_timestamp > $2)
+		`, ei.ID, time.Now().UTC())
+		if err != nil {
+			return fmt.Errorf("failed to list: %w", err)
+		}
+		defer rows.Close()
+
+		for rows.Next() {
+			if err := rows.Err(); err != nil {
+				return fmt.Errorf("failed to iterate: %w", err)
+			}
+
+			publicKey, err := scanOnePublicKey(rows)
+			if err != nil {
+				return err
+			}
+			publicKeys = append(publicKeys, publicKey)
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	return publicKeys, nil
+}
+
+func scanOnePublicKey(row pgx.Row) (*model.ImportFilePublicKey, error) {
+	var (
+		m    model.ImportFilePublicKey
+		thru *time.Time
+	)
+
+	if err := row.Scan(&m.ExportImportID, &m.KeyID, &m.KeyVersion, &m.PublicKeyPEM, &m.From, &thru); err != nil {
+		return nil, err
+	}
+	if thru != nil {
+		m.Thru = thru
+	}
+
+	return &m, nil
+}
+
+func prepareInsertImportFile(ctx context.Context, tx pgx.Tx) (string, error) {
+	const stmtName = "insert importfile"
+	_, err := tx.Prepare(ctx, stmtName, `
+		INSERT INTO
+			ImportFile
+				(export_import_id, zip_filename, discovered_at)
+		VALUES
+			($1, $2, $3)
+		RETURNING id
+	`)
+	return stmtName, err
+}
+
+func (db *ExportImportDB) CreateFiles(ctx context.Context, ei *model.ExportImport, filenames []string) (int, error) {
+	insertedFiles := 0
+
+	now := time.Now().UTC()
+	if err := db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		// read existing filenames
+		rows, err := tx.Query(ctx, `
+			SELECT
+				zip_filename
+			FROM
+				ImportFile
+			WHERE export_import_id=$1 AND zip_filename = ANY($2)
+		`, ei.ID, filenames)
+		if err != nil {
+			return fmt.Errorf("failed to read existing filenames: %w", err)
+		}
+		defer rows.Close()
+
+		// put existing ones into a map
+		existing := make(map[string]struct{})
+		for rows.Next() {
+			var fname string
+			if err := rows.Scan(&fname); err != nil {
+				return fmt.Errorf("failed to read existing filename: %w", err)
+			}
+			existing[fname] = struct{}{}
+		}
+
+		// Go through incoming list and insert a new entry for each filename we haven't seen before.
+		insertStmt, err := prepareInsertImportFile(ctx, tx)
+		if err != nil {
+			return fmt.Errorf("preparing insert statement: %v", err)
+		}
+
+		for _, fname := range filenames {
+			result, err := tx.Exec(ctx, insertStmt, ei.ID, fname, now)
+			if err != nil {
+				return fmt.Errorf("error inserting filename: %v, %w", fname, err)
+			}
+			if result.RowsAffected() != 1 {
+				return fmt.Errorf("filename isnert failed: %v", fname)
+			}
+			insertedFiles++
+		}
+		return nil
+	}); err != nil {
+		return 0, fmt.Errorf("creating import files: %w", err)
+	}
+
+	return insertedFiles, nil
+}
+
+func (db *ExportImportDB) CompleteImportFile(ctx context.Context, ef *model.ImportFile) error {
+	now := time.Now().UTC()
+	return db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		rows, err := tx.Query(ctx, `
+			SELECT
+				status
+			FROM
+				ImportFile
+			WHERE
+				id = $1
+			FOR UPDATE
+		`, ef.ID)
+		if err != nil {
+			return err
+		}
+		if !rows.Next() {
+			rows.Close()
+			return fmt.Errorf("record not found")
+		}
+		var curStatus string
+		if err := rows.Scan(&curStatus); err != nil {
+			rows.Close()
+			return fmt.Errorf("unable to read row: %w", err)
+		}
+
+		if curStatus != model.ImportFilePending {
+			rows.Close()
+			return fmt.Errorf("cannot complete from status: %v", curStatus)
+		}
+		rows.Close()
+
+		ef.Status = model.ImportFileComplete
+		ef.ProcessedAt = &now
+		result, err := tx.Exec(ctx, `
+			UPDATE
+				ImportFile
+			SET
+				status=$1, processed_at=$2
+			WHERE
+				id=$3
+			`, ef.Status, ef.ProcessedAt, ef.ID)
+		if err != nil {
+			return fmt.Errorf("unable to mark complete: %w", err)
+		}
+		if result.RowsAffected() != 1 {
+			return fmt.Errorf("marking complete did not change any rows")
+		}
+		return nil
+	})
+}
+
+func (db *ExportImportDB) LeaseImportFile(ctx context.Context, lockDuration time.Duration, ef *model.ImportFile) error {
+	now := time.Now().UTC().Truncate(time.Second)
+	return db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		rows, err := tx.Query(ctx, `
+			SELECT
+				status, processed_at
+			FROM
+				ImportFile
+			WHERE
+				id = $1
+			FOR UPDATE
+		`, ef.ID)
+		if err != nil {
+			return err
+		}
+		var curStatus string
+		var processedAt *time.Time
+		if !rows.Next() {
+			rows.Close()
+			return fmt.Errorf("record not found")
+		}
+		if err := rows.Scan(&curStatus, &processedAt); err != nil {
+			rows.Close()
+			return fmt.Errorf("unable to read row: %w", err)
+		}
+		rows.Close()
+
+		// A file that is open or was put into pending more than lockDuration ago can be locked.
+		okToLock := curStatus == model.ImportFileOpen ||
+			(curStatus == model.ImportFilePending &&
+				(processedAt == nil || now.Sub(*processedAt) > lockDuration))
+		if !okToLock {
+			return fmt.Errorf("file not elliglble for processing")
+		}
+
+		ef.Status = model.ImportFilePending
+		ef.ProcessedAt = &now
+		result, err := tx.Exec(ctx, `
+			UPDATE
+				ImportFile
+			SET
+				status=$1, processed_at=$2
+			WHERE
+				id=$3
+			`, ef.Status, ef.ProcessedAt, ef.ID)
+		if err != nil {
+			return fmt.Errorf("unable to lock file: %w", err)
+		}
+		if result.RowsAffected() != 1 {
+			return fmt.Errorf("row locking didn't update any rows")
+		}
+		return nil
+	})
+}
+
+func (db *ExportImportDB) GetOpenImportFiles(ctx context.Context, lockDuration time.Duration, ei *model.ExportImport) ([]*model.ImportFile, error) {
+	var importFiles []*model.ImportFile
+
+	lockOverrideTime := time.Now().UTC().Add(-1 * lockDuration)
+	if err := db.db.InTx(ctx, pgx.ReadCommitted, func(tx pgx.Tx) error {
+		rows, err := tx.Query(ctx, `
+			SELECT
+				id, zip_filename, discovered_at, status
+			FROM
+				ImportFile
+			WHERE
+				export_import_id = $1 AND (status = $2 OR (status = $3 AND processed_at < $4))
+			ORDER BY
+				id ASC
+		`, ei.ID, model.ImportFileOpen, model.ImportFilePending, lockOverrideTime)
+		if err != nil {
+			return fmt.Errorf("failed to list: %w", err)
+		}
+
+		defer rows.Close()
+		for rows.Next() {
+			if err := rows.Err(); err != nil {
+				return fmt.Errorf("failed to iterate: %w", err)
+			}
+
+			file := model.ImportFile{
+				ExportImportID: ei.ID,
+			}
+			if err := rows.Scan(&file.ID, &file.ZipFilename, &file.DiscoveredAt, &file.Status); err != nil {
+				return err
+			}
+			importFiles = append(importFiles, &file)
+		}
+		return nil
+	}); err != nil {
+		return nil, fmt.Errorf("unable to read open import files: %w", err)
+	}
+
+	return importFiles, nil
+}
diff --git a/internal/exportimport/import.go b/internal/exportimport/import.go
new file mode 100644
index 0000000..b1ddc5f
--- /dev/null
+++ b/internal/exportimport/import.go
@@ -0,0 +1,215 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package exportimport imports export files into the local database
+package exportimport
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"time"
+
+	"github.com/google/exposure-notifications-server/internal/export"
+	"github.com/google/exposure-notifications-server/internal/exportimport/model"
+	exportproto "github.com/google/exposure-notifications-server/internal/pb/export"
+	pubdb "github.com/google/exposure-notifications-server/internal/publish/database"
+	pubmodel "github.com/google/exposure-notifications-server/internal/publish/model"
+	"github.com/google/exposure-notifications-server/pkg/logging"
+	"go.uber.org/zap"
+)
+
+type ImportRequest struct {
+	config       *Config
+	exportImport *model.ExportImport
+	keys         []*model.ImportFilePublicKey
+	file         *model.ImportFile
+}
+
+type ImportResposne struct {
+	insertedKeys uint64
+	revisedKeys  uint64
+	droppedKeys  uint64
+}
+
+type SignatureAndKey struct {
+	signature []byte
+	publicKey *ecdsa.PublicKey
+}
+
+func (s *Server) ImportExportFile(ctx context.Context, ir *ImportRequest) (*ImportResposne, error) {
+	logger := logging.FromContext(ctx)
+	// Download zip file.
+	client := &http.Client{
+		Timeout: s.config.ExportFileDownloadTimeout,
+	}
+	resp, err := client.Get(ir.file.ZipFilename)
+	if err != nil {
+		return nil, fmt.Errorf("error downloading export file: %w", err)
+	}
+
+	defer resp.Body.Close()
+	bytes, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("error reading file: %w", err)
+	}
+
+	// Get bin and sig files.
+	tekExport, digest, err := export.UnmarshalExportFile(bytes)
+	if err != nil {
+		return nil, fmt.Errorf("bin data error: %w", err)
+	}
+	tekSignatures, err := export.UnmarshalSignatureFile(bytes)
+	if err != nil {
+		return nil, fmt.Errorf("signature data missing: %w", err)
+	}
+
+	// Index the signatures from the file.
+	signatures := make(map[string]*SignatureAndKey)
+	for _, tekSig := range tekSignatures.GetSignatures() {
+		idAndVersion := fmt.Sprintf("%s.%s", tekSig.SignatureInfo.GetVerificationKeyId(), tekSig.SignatureInfo.GetVerificationKeyVersion())
+		signatures[idAndVersion] = &SignatureAndKey{
+			signature: tekSig.GetSignature(),
+		}
+	}
+	// Join in available public keys
+	for _, key := range ir.keys {
+		idAndVersion := fmt.Sprintf("%s.%s", key.KeyID, key.KeyVersion)
+		if sak, ok := signatures[idAndVersion]; ok {
+			sak.publicKey, err = key.PublicKey()
+			if err != nil {
+				return nil, fmt.Errorf("unable to parse public key: %w", err)
+			}
+		} else {
+			logger.Infow("key not found...", "idAndVersion", idAndVersion)
+		}
+	}
+
+	// Validate signatures.
+	valid := false
+	for k, sig := range signatures {
+		if sig.publicKey == nil {
+			logger.Warnw("no public key for signature", "signature", sig)
+			continue
+		}
+		if ecdsa.VerifyASN1(sig.publicKey, digest[:], sig.signature) {
+			valid = true
+			logger.Debugw("validated signature", "file", ir.file, "kid.version", k)
+			break
+		}
+	}
+	if !valid {
+		return nil, fmt.Errorf("no valid signature found")
+	}
+
+	// Common transform settings for primar + revised keys.
+	exKeyTransform := transformer{
+		appPackageName:               s.config.ImportAPKName,
+		importRegions:                []string{ir.exportImport.Region},
+		batchTime:                    time.Now().UTC().Truncate(s.config.CreatedAtTruncateWindow),
+		exportImportID:               ir.exportImport.ID,
+		importFileID:                 ir.file.ID,
+		maxMagnitudeSymptomOnsetDays: int32(s.config.MaxMagnitudeSymptomOnsetDays),
+		logger:                       logger,
+	}
+	response := ImportResposne{}
+
+	// Go through primary keys and insert.
+	// Must be separate from revised keys in the event both are in the same file.
+	if len(tekExport.Keys) > 0 {
+		inserts, dropped := exKeyTransform.transform(tekExport.Keys)
+		response.droppedKeys = response.droppedKeys + dropped
+		template := pubdb.InsertAndReviseExposuresRequest{
+			SkipRevions: true,
+		}
+		if err := s.insertAndReviseKeys(ctx, "primary", inserts, &template, &response); err != nil {
+			return nil, fmt.Errorf("insert primary keys: %w", err)
+		}
+	}
+
+	// Then revised keys and revise.
+	if len(tekExport.RevisedKeys) > 0 {
+		revisions, dropped := exKeyTransform.transform(tekExport.RevisedKeys)
+		response.droppedKeys = response.droppedKeys + dropped
+		template := pubdb.InsertAndReviseExposuresRequest{
+			OnlyRevisions:         true,
+			RequireToken:          false,
+			RequireExportImportID: true,
+		}
+		if err := s.insertAndReviseKeys(ctx, "revised", revisions, &template, &response); err != nil {
+			return nil, fmt.Errorf("insert revised keys: %w", err)
+		}
+	}
+
+	return &response, nil
+}
+
+func (s *Server) insertAndReviseKeys(ctx context.Context, mode string, exposures []*pubmodel.Exposure, template *pubdb.InsertAndReviseExposuresRequest, response *ImportResposne) error {
+	logger := logging.FromContext(ctx)
+	length := len(exposures)
+	for i := 0; i < length; i = i + s.config.MaxInsertBatchSize {
+		upper := i + s.config.MaxInsertBatchSize
+		if upper > length {
+			upper = length
+		}
+		// Assign the current operating slice.
+		template.Incoming = exposures[i:upper]
+
+		insertResponse, err := s.publishDB.InsertAndReviseExposures(ctx, template)
+		if err != nil {
+			return fmt.Errorf("publishDB.InsertAndReviseExposures: %w", err)
+		}
+		logger.Infow("insertAndRevise", "mode", mode, "candidates", len(template.Incoming), "inserted", insertResponse.Inserted, "revised", insertResponse.Revised, "dropped", insertResponse.Dropped)
+
+		response.insertedKeys = response.insertedKeys + insertResponse.Inserted
+		response.droppedKeys = response.droppedKeys + insertResponse.Dropped
+		response.revisedKeys = response.revisedKeys + insertResponse.Revised
+	}
+	return nil
+}
+
+type transformer struct {
+	appPackageName               string
+	importRegions                []string
+	batchTime                    time.Time
+	exportImportID               int64
+	importFileID                 int64
+	maxMagnitudeSymptomOnsetDays int32
+	logger                       *zap.SugaredLogger
+}
+
+func (t *transformer) transform(keys []*exportproto.TemporaryExposureKey) ([]*pubmodel.Exposure, uint64) {
+	inserts := make([]*pubmodel.Exposure, 0, len(keys))
+	var dropped uint64
+	for _, k := range keys {
+		exp, err := pubmodel.FromExportKey(k, t.maxMagnitudeSymptomOnsetDays)
+		if err != nil {
+			t.logger.Warnw("skipping invalid key", "error", err)
+			dropped++
+			continue
+		}
+		// Fill in items that are specific to this import.
+		exp.AppPackageName = t.appPackageName
+		exp.Regions = t.importRegions
+		exp.CreatedAt = t.batchTime
+		exp.LocalProvenance = false
+		exp.ExportImportID = &t.exportImportID
+		exp.ImportFileID = &t.importFileID
+
+		inserts = append(inserts, exp)
+	}
+	return inserts, dropped
+}
diff --git a/internal/exportimport/importer.go b/internal/exportimport/importer.go
new file mode 100644
index 0000000..53f3505
--- /dev/null
+++ b/internal/exportimport/importer.go
@@ -0,0 +1,134 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exportimport
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/pkg/logging"
+	"go.opencensus.io/trace"
+)
+
+const lockPrefix = "import-lock-"
+
+func (s *Server) handleImport(ctx context.Context) http.HandlerFunc {
+	logger := logging.FromContext(ctx).Named("exportimport.HandleImport")
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		_, span := trace.StartSpan(r.Context(), "(*keyrotation.handler).ServeHTTP")
+		defer span.End()
+
+		ctx, cancelFn := context.WithDeadline(r.Context(), time.Now().Add(s.config.MaxRuntime))
+		defer cancelFn()
+		logger.Info("starting export importer")
+		defer func() {
+			w.WriteHeader(http.StatusOK)
+			w.Write([]byte("ok"))
+		}()
+
+		configs, err := s.exportImportDB.ActiveConfigs(ctx)
+		if err != nil {
+			logger.Errorw("unable to read active configs", "error", err)
+		}
+
+		for _, config := range configs {
+			// Check how we're doing on max runtime.
+			if deadlinePassed(ctx) {
+				logger.Warnf("deadline passed, still work to do")
+				return
+			}
+
+			// Obtain a lock to work on this import config.
+			unlock, err := s.db.Lock(ctx, fmt.Sprintf("%s%d", lockPrefix, config.ID), s.config.MaxRuntime)
+			if err != nil {
+				if errors.Is(err, database.ErrAlreadyLocked) {
+					logger.Warnw("import already locked", "config", config)
+				}
+				logger.Errorw("error locking import config", "config", config, "error", err)
+				continue
+			}
+			defer func() {
+				if err := unlock(); err != nil {
+					logger.Errorf("failed to unlock: %v", err)
+				}
+			}()
+
+			// Get the list of files that needs to be processed.
+			openFiles, err := s.exportImportDB.GetOpenImportFiles(ctx, s.config.ImportLockTime, config)
+			if err != nil {
+				logger.Errorw("unable to read open export files", "config", config, "error", err)
+			}
+			if len(openFiles) == 0 {
+				logger.Infow("no work to do", "config", config)
+				continue
+			}
+
+			// Read in public keys.
+			keys, err := s.exportImportDB.AllowedKeys(ctx, config)
+			if err != nil {
+				logger.Errorw("unable to read public keys", "config", config, "error", err)
+				continue
+			}
+			logger.Debugw("allowed public keys for file", "publicKeys", keys)
+
+			for _, file := range openFiles {
+				// Check how we're doing on max runtime.
+				if deadlinePassed(ctx) {
+					logger.Warnw("deadline passed, still work to do", "config", config)
+					return
+				}
+
+				if err := s.exportImportDB.LeaseImportFile(ctx, s.config.ImportLockTime, file); err != nil {
+					logger.Warnw("unexpected race condition, file already locked", "file", file, "error", err)
+					continue
+				}
+
+				// import the file.
+				result, err := s.ImportExportFile(ctx, &ImportRequest{
+					config:       s.config,
+					exportImport: config,
+					keys:         keys,
+					file:         file,
+				})
+				if err != nil {
+					logger.Errorf("error processing import file", "error", err)
+					continue
+				}
+				logger.Infow("completed file import", "inserted", result.insertedKeys, "revised", result.revisedKeys, "dropped", result.droppedKeys)
+
+				if err := s.exportImportDB.CompleteImportFile(ctx, file); err != nil {
+					logger.Errorf("failed to mark file completed", "file", file, "error", err)
+				}
+			}
+		}
+
+	}
+}
+
+func deadlinePassed(ctx context.Context) bool {
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		return false
+	}
+	if time.Now().After(deadline) {
+		return true
+	}
+	return false
+}
diff --git a/internal/exportimport/model/exportimport.go b/internal/exportimport/model/exportimport.go
new file mode 100644
index 0000000..cd61596
--- /dev/null
+++ b/internal/exportimport/model/exportimport.go
@@ -0,0 +1,52 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package model is a model abstraction of export import configuration and status.
+package model
+
+import (
+	"fmt"
+	"time"
+)
+
+type ExportImport struct {
+	ID         int64
+	IndexFile  string
+	ExportRoot string
+	Region     string
+	From       time.Time
+	Thru       *time.Time
+}
+
+func (ei *ExportImport) Validate() error {
+	if l := len(ei.Region); l == 0 {
+		return fmt.Errorf("region cannot be blank")
+	} else if l > 5 {
+		return fmt.Errorf("region cannot be longer than 5 characters")
+	}
+
+	if ei.IndexFile == "" {
+		return fmt.Errorf("IndexFile cannot be blank")
+	}
+	if ei.ExportRoot == "" {
+		return fmt.Errorf("ExportRoot cannot be blank")
+	}
+
+	return nil
+}
+
+func (ei *ExportImport) Active() bool {
+	now := time.Now().UTC()
+	return ei.From.Before(now) && (ei.Thru == nil || now.Before(*ei.Thru))
+}
diff --git a/internal/exportimport/model/file.go b/internal/exportimport/model/file.go
new file mode 100644
index 0000000..8bce21a
--- /dev/null
+++ b/internal/exportimport/model/file.go
@@ -0,0 +1,33 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package model
+
+import "time"
+
+const (
+	ImportFileOpen     = "OPEN"
+	ImportFilePending  = "PENDING"
+	ImportFileComplete = "COMPLETE"
+	ImportFileFailed   = "FAILED"
+)
+
+type ImportFile struct {
+	ID             int64
+	ExportImportID int64
+	ZipFilename    string
+	DiscoveredAt   time.Time
+	ProcessedAt    *time.Time
+	Status         string
+}
diff --git a/internal/exportimport/model/importkeys.go b/internal/exportimport/model/importkeys.go
new file mode 100644
index 0000000..24a2cb1
--- /dev/null
+++ b/internal/exportimport/model/importkeys.go
@@ -0,0 +1,51 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package model
+
+import (
+	"crypto/ecdsa"
+	"crypto/x509"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"time"
+)
+
+type ImportFilePublicKey struct {
+	ExportImportID int64
+	KeyID          string
+	KeyVersion     string
+	PublicKeyPEM   string
+	From           time.Time
+	Thru           *time.Time
+}
+
+func (pk *ImportFilePublicKey) PublicKey() (*ecdsa.PublicKey, error) {
+	block, _ := pem.Decode([]byte(pk.PublicKeyPEM))
+	if block == nil {
+		return nil, errors.New("unable to decode PEM block containing PUBLIC KEY")
+	}
+	pub, err := x509.ParsePKIXPublicKey(block.Bytes)
+	if err != nil {
+		return nil, fmt.Errorf("x509.ParsePKIXPublicKey: %w", err)
+	}
+
+	switch typ := pub.(type) {
+	case *ecdsa.PublicKey:
+		return typ, nil
+	default:
+		return nil, fmt.Errorf("unsupported public key type: %T", typ)
+	}
+}
diff --git a/internal/exportimport/scheduler.go b/internal/exportimport/scheduler.go
new file mode 100644
index 0000000..2728417
--- /dev/null
+++ b/internal/exportimport/scheduler.go
@@ -0,0 +1,107 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exportimport
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/google/exposure-notifications-server/pkg/logging"
+	"go.opencensus.io/trace"
+)
+
+const schedulerLockID = "import-scheduler-lock"
+
+func (s *Server) handleSchedule(ctx context.Context) http.HandlerFunc {
+	logger := logging.FromContext(ctx).Named("exportimport.HandleSchedule")
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		_, span := trace.StartSpan(r.Context(), "(*exportimport.handleSchedule).ServeHTTP")
+		defer span.End()
+
+		unlock, err := s.db.Lock(ctx, schedulerLockID, s.config.MaxRuntime)
+		if err != nil {
+			logger.Warn(err)
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		defer func() {
+			if err := unlock(); err != nil {
+				logger.Errorf("failed to unlock: %v", err)
+			}
+		}()
+
+		ctx, cancelFn := context.WithDeadline(r.Context(), time.Now().Add(s.config.MaxRuntime))
+		defer cancelFn()
+		logger.Info("starting export import scheduler")
+
+		configs, err := s.exportImportDB.ActiveConfigs(ctx)
+		if err != nil {
+			logger.Errorw("unable to read active configs", "error", err)
+			return
+		}
+
+		client := &http.Client{
+			Timeout: s.config.IndexFileDownloadTimeout,
+		}
+
+		anyErrors := false
+		for _, config := range configs {
+			logger.Infow("checking index file", "config", config)
+
+			resp, err := client.Get(config.IndexFile)
+			if err != nil {
+				anyErrors = true
+				logger.Errorw("error downloading index file", "file", config.IndexFile, "error", err)
+				continue
+			}
+
+			defer resp.Body.Close()
+			bytes, err := ioutil.ReadAll(resp.Body)
+			if err != nil {
+				anyErrors = true
+				logger.Errorw("unable to read index file", "file", config.IndexFile, "error", err)
+				continue
+			}
+			zipNames := strings.Split(string(bytes), "\n")
+			currentFiles := make([]string, 0, len(zipNames))
+			for _, zipFile := range zipNames {
+				fullZipFile := fmt.Sprintf("%s%s", config.ExportRoot, strings.TrimSpace(zipFile))
+				logger.Debugw("found new export file", "zipFile", fullZipFile)
+				currentFiles = append(currentFiles, fullZipFile)
+			}
+
+			n, err := s.exportImportDB.CreateFiles(ctx, config, currentFiles)
+			if err != nil {
+				anyErrors = true
+				logger.Errorw("unable to write new files", "error", err)
+			}
+			if n != 0 {
+				logger.Infow("found new files for export import config", "file", config.IndexFile, "newCount", n)
+			}
+		}
+
+		status := http.StatusOK
+		if anyErrors {
+			status = http.StatusInternalServerError
+		}
+		w.WriteHeader(status)
+		w.Write([]byte(http.StatusText(status)))
+	}
+}
diff --git a/internal/exportimport/server.go b/internal/exportimport/server.go
new file mode 100644
index 0000000..703afe0
--- /dev/null
+++ b/internal/exportimport/server.go
@@ -0,0 +1,68 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package exportimport implements the handlers for the export-importer functionality.
+package exportimport
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/google/exposure-notifications-server/internal/database"
+	eidb "github.com/google/exposure-notifications-server/internal/exportimport/database"
+	pubdb "github.com/google/exposure-notifications-server/internal/publish/database"
+	"github.com/google/exposure-notifications-server/internal/serverenv"
+	"github.com/google/exposure-notifications-server/pkg/server"
+)
+
+// Server hosts end points to manage key rotation
+type Server struct {
+	config         *Config
+	env            *serverenv.ServerEnv
+	db             *database.DB
+	exportImportDB *eidb.ExportImportDB
+	publishDB      *pubdb.PublishDB
+}
+
+// NewServer creates a Server that manages deletion of
+// old export files that are no longer needed by clients for download.
+func NewServer(config *Config, env *serverenv.ServerEnv) (*Server, error) {
+	if env.Database() == nil {
+		return nil, fmt.Errorf("missing database in server environment")
+	}
+
+	db := env.Database()
+	exportImportDB := eidb.New(db)
+	publishDB := pubdb.New(db)
+
+	return &Server{
+		config:         config,
+		env:            env,
+		db:             db,
+		exportImportDB: exportImportDB,
+		publishDB:      publishDB,
+	}, nil
+}
+
+// Routes defines and returns the routes for this server.
+func (s *Server) Routes(ctx context.Context) *http.ServeMux {
+	mux := http.NewServeMux()
+
+	mux.HandleFunc("/schedule", s.handleSchedule(ctx))
+	mux.HandleFunc("/import", s.handleImport(ctx))
+	mux.Handle("/health", server.HandleHealthz(ctx))
+
+	return mux
+}
diff --git a/internal/publish/database/exposure.go b/internal/publish/database/exposure.go
index bb853c2..eb14437 100644
--- a/internal/publish/database/exposure.go
+++ b/internal/publish/database/exposure.go
@@ -274,7 +274,7 @@ func (db *PublishDB) ReadExposures(ctx context.Context, tx pgx.Tx, b64keys []str
 				interval_number, interval_count, created_at, local_provenance, sync_id,
 				health_authority_id, report_type, days_since_symptom_onset,
 				revised_report_type, revised_at, revised_days_since_symptom_onset,
-				revised_transmission_risk
+				revised_transmission_risk, export_import_id
 			FROM
 				Exposure
 			WHERE exposure_key = ANY($1)
@@ -300,7 +300,7 @@ func (db *PublishDB) ReadExposures(ctx context.Context, tx pgx.Tx, b64keys []str
 				&exposure.CreatedAt, &exposure.LocalProvenance, &syncID,
 				&exposure.HealthAuthorityID, &exposure.ReportType, &exposure.DaysSinceSymptomOnset,
 				&exposure.RevisedReportType, &exposure.RevisedAt, &exposure.RevisedDaysSinceSymptomOnset,
-				&exposure.RevisedTransmissionRisk,
+				&exposure.RevisedTransmissionRisk, &exposure.ExportImportID,
 			); err != nil {
 				return fmt.Errorf("failed to parse: %w", err)
 			}
@@ -332,9 +332,10 @@ func prepareInsertExposure(ctx context.Context, tx pgx.Tx) (string, error) {
 		INSERT INTO
 			Exposure
 				(exposure_key, transmission_risk, app_package_name, regions, traveler, interval_number, interval_count,
-				 created_at, local_provenance, sync_id, sync_query_id, health_authority_id, report_type, days_since_symptom_onset)
+				 created_at, local_provenance, sync_id, sync_query_id, health_authority_id, report_type, days_since_symptom_onset,
+				 export_import_id, import_file_id)
 		VALUES
-			($1, $2, LOWER($3), $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
+			($1, $2, LOWER($3), $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
 		ON CONFLICT (exposure_key) DO NOTHING
 	`)
 	return stmtName, err
@@ -353,7 +354,8 @@ func executeInsertExposure(ctx context.Context, tx pgx.Tx, stmtName string, exp
 	_, err := tx.Exec(ctx, stmtName, encodeExposureKey(exp.ExposureKey), exp.TransmissionRisk,
 		exp.AppPackageName, exp.Regions, exp.Traveler, exp.IntervalNumber, exp.IntervalCount,
 		exp.CreatedAt, exp.LocalProvenance, syncID, queryID,
-		exp.HealthAuthorityID, exp.ReportType, exp.DaysSinceSymptomOnset)
+		exp.HealthAuthorityID, exp.ReportType, exp.DaysSinceSymptomOnset,
+		exp.ExportImportID, exp.ImportFileID)
 	if err != nil {
 		return fmt.Errorf("inserting exposure: %v", err)
 	}
@@ -367,9 +369,10 @@ func prepareReviseExposure(ctx context.Context, tx pgx.Tx) (string, error) {
 			Exposure
 		SET
 			health_authority_id = $1, revised_report_type = $2, revised_at = $3,
-			revised_days_since_symptom_onset = $4, revised_transmission_risk = $5
+			revised_days_since_symptom_onset = $4, revised_transmission_risk = $5,
+			revised_import_file_id = $6
 		WHERE
-			exposure_key = $6 AND revised_at IS NULL
+			exposure_key = $7 AND revised_at IS NULL
 		`)
 	return stmtName, err
 }
@@ -378,6 +381,7 @@ func executeReviseExposure(ctx context.Context, tx pgx.Tx, stmtName string, exp
 	result, err := tx.Exec(ctx, stmtName,
 		exp.HealthAuthorityID, exp.RevisedReportType, exp.RevisedAt,
 		exp.RevisedDaysSinceSymptomOnset, exp.RevisedTransmissionRisk,
+		exp.RevisedImportFileID,
 		encodeExposureKey(exp.ExposureKey))
 	if err != nil {
 		return fmt.Errorf("revising exposure: %v", err)
@@ -410,6 +414,8 @@ type InsertAndReviseExposuresRequest struct {
 	OnlyRevisions bool
 	// Require matching Sync QueryID only allows revisions if they originated from the same query ID.
 	RequireQueryID bool
+	// When revising, require matching export-import-ID. For export file based import federation.
+	RequireExportImportID bool
 }
 
 // InsertAndReviseExposuresResponse is the response from an
@@ -485,8 +491,8 @@ func (db *PublishDB) InsertAndReviseExposures(ctx context.Context, req *InsertAn
 		if len(existing) > 0 {
 			// Check if a revision token is required.
 			if req.Token == nil {
-				logger.Errorw("attempted to revise keys, but revision token is missing")
 				if req.RequireToken {
+					logger.Warnw("attempted to revise keys, but revision token is missing")
 					return ErrNoRevisionToken
 				}
 			}
@@ -520,6 +526,18 @@ func (db *PublishDB) InsertAndReviseExposures(ctx context.Context, req *InsertAn
 						}
 					}
 				}
+				// For export file based federation. Revisions must come from the same export lineage.
+				if req.RequireExportImportID {
+					if in, ok := incomingMap[k]; ok {
+						inID := in.ExportImportID
+						pID := ex.ExportImportID
+						if inID == nil || pID == nil || *inID != *pID {
+							logger.Warnw("key revision attempted on import with wrong origin", "exportImportID", ex.ExportImportID, "proposedexportImportID", in.ExportImportID)
+							delete(incomingMap, k)
+							continue
+						}
+					}
+				}
 
 				// Check the incoming values first.
 				if in, ok := incomingMap[k]; ok {
diff --git a/internal/publish/model/exposure_model.go b/internal/publish/model/exposure_model.go
index e8402b8..acde1d4 100644
--- a/internal/publish/model/exposure_model.go
+++ b/internal/publish/model/exposure_model.go
@@ -24,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/google/exposure-notifications-server/internal/pb/export"
 	"github.com/google/exposure-notifications-server/internal/verification"
 	"github.com/google/exposure-notifications-server/pkg/base64util"
 	"github.com/google/exposure-notifications-server/pkg/logging"
@@ -70,6 +71,10 @@ type Exposure struct {
 	LocalProvenance   bool
 	FederationSyncID  int64
 	FederationQueryID string
+	// Export file based Federation
+	ExportImportID      *int64
+	ImportFileID        *int64
+	RevisedImportFileID *int64
 
 	// These fields are nullable to maintain backwards compatibility with
 	// older versions that predate their existence.
@@ -87,6 +92,77 @@ type Exposure struct {
 	base64Key string
 }
 
+// FromExportKey is used to read a key from an export file and convert it back to the
+// internal database format.
+func FromExportKey(key *export.TemporaryExposureKey, maxSymptomOnsetDays int32) (*Exposure, error) {
+	exp := &Exposure{
+		ExposureKey: make([]byte, verifyapi.KeyLength),
+	}
+	if len(key.KeyData) != verifyapi.KeyLength {
+		return nil, fmt.Errorf("invalid key length")
+	}
+	copy(exp.ExposureKey, key.KeyData)
+
+	//lint:ignore SA1019 may be set on v1 files.
+	if key.TransmissionRiskLevel != nil {
+		//lint:ignore SA1019 may be set on v1 files.
+		if tr := *key.TransmissionRiskLevel; tr < verifyapi.MinTransmissionRisk {
+			return nil, fmt.Errorf("transmission risk too low: %d, must be >= %d", tr, verifyapi.MinTransmissionRisk)
+		} else if tr > verifyapi.MaxTransmissionRisk {
+			return nil, fmt.Errorf("transmission risk too high: %d, must be <= %d", tr, verifyapi.MaxTransmissionRisk)
+		} else {
+			exp.TransmissionRisk = int(tr)
+		}
+	}
+
+	if key.RollingStartIntervalNumber == nil {
+		return nil, fmt.Errorf("missing rolling_start_interval_number")
+	}
+	exp.IntervalNumber = *key.RollingStartIntervalNumber
+
+	if key.RollingPeriod == nil {
+		exp.IntervalCount = verifyapi.MaxIntervalCount
+	} else {
+		if rp := *key.RollingPeriod; rp < verifyapi.MinIntervalCount {
+			return nil, fmt.Errorf("rolling period too low: %d must be >= %d", rp, verifyapi.MinIntervalCount)
+		} else if rp > verifyapi.MaxIntervalCount {
+			return nil, fmt.Errorf("rolling period too low: %d must be <= %d", rp, verifyapi.MaxIntervalCount)
+		} else {
+			exp.IntervalCount = rp
+		}
+	}
+
+	if key.ReportType != nil {
+		rt := *key.ReportType
+		switch rt {
+		case export.TemporaryExposureKey_CONFIRMED_TEST:
+			exp.ReportType = verifyapi.ReportTypeConfirmed
+		case export.TemporaryExposureKey_CONFIRMED_CLINICAL_DIAGNOSIS:
+			exp.ReportType = verifyapi.ReportTypeClinical
+		case export.TemporaryExposureKey_REVOKED:
+			exp.ReportType = verifyapi.ReportTypeNegative
+		case export.TemporaryExposureKey_UNKNOWN:
+			exp.ReportType = ""
+		default:
+			return nil, fmt.Errorf("unsupported report type: %s",
+				export.TemporaryExposureKey_ReportType_name[int32(rt)])
+		}
+	}
+
+	if key.DaysSinceOnsetOfSymptoms != nil {
+		dsos := *key.DaysSinceOnsetOfSymptoms
+		if dsos >= (-1*maxSymptomOnsetDays) && dsos <= maxSymptomOnsetDays {
+			exp.DaysSinceSymptomOnset = &dsos
+		} else {
+			return nil, fmt.Errorf("days since onset of symptoms is out of range: %d must be within: %d", dsos, maxSymptomOnsetDays)
+		}
+	}
+
+	exp.LocalProvenance = false
+
+	return exp, nil
+}
+
 // Revise updates the Revised fields of a key
 func (e *Exposure) Revise(in *Exposure) (bool, error) {
 	if e.ExposureKeyBase64() != in.ExposureKeyBase64() {
@@ -97,10 +173,21 @@ func (e *Exposure) Revise(in *Exposure) (bool, error) {
 		return false, nil
 	}
 	if !e.LocalProvenance {
-		if e.FederationQueryID == "" {
+		nonLocalOK := false
+		if e.ExportImportID != nil {
+			if in.ExportImportID == nil || *e.ExportImportID != *in.ExportImportID {
+				return false, ErrorNotSameFederationSource
+			}
+			nonLocalOK = true
+		} else if e.FederationQueryID != "" {
+			if e.FederationQueryID != in.FederationQueryID {
+				return false, ErrorNotSameFederationSource
+			}
+			nonLocalOK = true
+		}
+
+		if !nonLocalOK {
 			return false, ErrorNonLocalProvenance
-		} else if e.FederationQueryID != in.FederationQueryID {
-			return false, ErrorNotSameFederationSource
 		}
 	}
 	// make sure key hasn't been revised already.
@@ -132,6 +219,9 @@ func (e *Exposure) Revise(in *Exposure) (bool, error) {
 	tr := ReportTypeTransmissionRisk(in.ReportType, in.TransmissionRisk)
 	e.RevisedTransmissionRisk = &tr
 
+	e.HealthAuthorityID = in.HealthAuthorityID
+	e.RevisedImportFileID = in.ImportFileID
+
 	return true, nil
 }
 
diff --git a/migrations/000042_ReAddExportImporterTables.down.sql b/migrations/000042_ReAddExportImporterTables.down.sql
new file mode 100644
index 0000000..dc1f191
--- /dev/null
+++ b/migrations/000042_ReAddExportImporterTables.down.sql
@@ -0,0 +1,29 @@
+-- Copyright 2020 Google LLC
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+BEGIN;
+
+ALTER TABLE Exposure
+    DROP COLUMN export_import_id,
+    DROP COLUMN import_file_id,
+    DROP COLUMN revised_import_file_id;
+
+DROP TABLE ImportFile;
+DROP TYPE ImportFileStatus;
+
+DROP TABLE ImportFilePublicKey;
+
+DROP TABLE ExportImport;
+
+END;
diff --git a/migrations/000042_ReAddExportImporterTables.up.sql b/migrations/000042_ReAddExportImporterTables.up.sql
new file mode 100644
index 0000000..f0f13eb
--- /dev/null
+++ b/migrations/000042_ReAddExportImporterTables.up.sql
@@ -0,0 +1,67 @@
+-- Copyright 2020 Google LLC
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--      http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+BEGIN;
+
+-- Top level configuration for other key server export files
+-- that should be imported into this server.
+CREATE TABLE ExportImport(
+    id SERIAL PRIMARY KEY,
+    index_file VARCHAR(500) NOT NULL,
+    export_root VARCHAR(500) NOT NULL,
+    region VARCHAR(5) NOT NULL,
+    from_timestamp TIMESTAMPTZ NOT NULL,
+	thru_timestamp TIMESTAMPTZ
+);
+
+CREATE INDEX export_import_from_timestamp ON ExportImport USING BRIN(from_timestamp);
+CREATE INDEX export_import_thru_timestamp ON ExportImport USING BRIN(thru_timestamp);
+
+-- Public key information for an import config.
+CREATE TABLE ImportFilePublicKey(
+    export_import_id INT NOT NULL REFERENCES ExportImport(id),
+    key_id VARCHAR(50) NOT NULL,
+    key_version VARCHAR(50) NOT NULL,
+    public_key VARCHAR(500) NOT NULL,
+    from_timestamp TIMESTAMPTZ NOT NULL,
+	thru_timestamp TIMESTAMPTZ,
+    PRIMARY KEY(export_import_id, key_id, key_version)
+);
+
+
+CREATE INDEX import_file_public_key_from ON ImportFilePublicKey USING BRIN(from_timestamp);
+CREATE INDEX import_file_public_key_thru ON ImportFilePublicKey USING BRIN(thru_timestamp);
+
+-- Individual .zip export files that are being imported into this serever.
+CREATE TYPE ImportFileStatus AS ENUM ('OPEN', 'PENDING', 'COMPLETE', 'FAILED');
+CREATE TABLE ImportFile (
+    id SERIAL PRIMARY KEY,
+    export_import_id INT NOT NULL REFERENCES ExportImport(id),
+    zip_filename VARCHAR(500) NOT NULL,
+    discovered_at TIMESTAMPTZ NOT NULL,
+    processed_at TIMESTAMPTZ,
+    status ImportFileStatus NOT NULL DEFAULT 'OPEN' 
+);
+
+CREATE INDEX import_file_status_processed ON ImportFile (status, processed_at);
+
+-- Allows us to link where an imported TEK came from
+-- and marks when we saw the revision as well.
+ALTER TABLE Exposure
+    ADD COLUMN export_import_id INT,
+    ADD COLUMN import_file_id INT,
+    ADD COLUMN revised_import_file_id INT;
+
+
+END;
diff --git a/terraform/service_cleanup_export.tf b/terraform/service_cleanup_export.tf
index ae983aa..bf22d8a 100644
--- a/terraform/service_cleanup_export.tf
+++ b/terraform/service_cleanup_export.tf
@@ -112,6 +112,7 @@ resource "google_cloud_run_service" "cleanup-export" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.cleanup-export-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_cleanup_exposure.tf b/terraform/service_cleanup_exposure.tf
index 77b6ddc..79e7900 100644
--- a/terraform/service_cleanup_exposure.tf
+++ b/terraform/service_cleanup_exposure.tf
@@ -106,6 +106,7 @@ resource "google_cloud_run_service" "cleanup-exposure" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.cleanup-exposure-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_debugger.tf b/terraform/service_debugger.tf
index e2b73b9..5d07da6 100644
--- a/terraform/service_debugger.tf
+++ b/terraform/service_debugger.tf
@@ -116,6 +116,7 @@ resource "google_cloud_run_service" "debugger" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.debugger-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_export.tf b/terraform/service_export.tf
index b64a31a..09a68cf 100644
--- a/terraform/service_export.tf
+++ b/terraform/service_export.tf
@@ -120,6 +120,7 @@ resource "google_cloud_run_service" "export" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.export-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_exposure.tf b/terraform/service_exposure.tf
index 40530b7..028f92a 100644
--- a/terraform/service_exposure.tf
+++ b/terraform/service_exposure.tf
@@ -122,6 +122,7 @@ resource "google_cloud_run_service" "exposure" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.exposure-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_federationin.tf b/terraform/service_federationin.tf
index 82d9f80..fbbb4a8 100644
--- a/terraform/service_federationin.tf
+++ b/terraform/service_federationin.tf
@@ -106,6 +106,7 @@ resource "google_cloud_run_service" "federationin" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.federationin,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_federationout.tf b/terraform/service_federationout.tf
index 17456a9..a02bcad 100644
--- a/terraform/service_federationout.tf
+++ b/terraform/service_federationout.tf
@@ -106,6 +106,7 @@ resource "google_cloud_run_service" "federationout" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.federationout-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_generate.tf b/terraform/service_generate.tf
index 5fe8e8e..7298b49 100644
--- a/terraform/service_generate.tf
+++ b/terraform/service_generate.tf
@@ -106,6 +106,7 @@ resource "google_cloud_run_service" "generate" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.generate-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_jwks.tf b/terraform/service_jwks.tf
index 60a1092..ace2020 100644
--- a/terraform/service_jwks.tf
+++ b/terraform/service_jwks.tf
@@ -106,6 +106,7 @@ resource "google_cloud_run_service" "jwks" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.jwks-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/terraform/service_keyrotation.tf b/terraform/service_keyrotation.tf
index 8301af1..57b6895 100644
--- a/terraform/service_keyrotation.tf
+++ b/terraform/service_keyrotation.tf
@@ -122,6 +122,7 @@ resource "google_cloud_run_service" "key-rotation" {
     google_project_service.services["run.googleapis.com"],
     google_secret_manager_secret_iam_member.key-rotation-db,
     null_resource.build,
+    null_resource.migrate,
   ]
 
   lifecycle {
diff --git a/tools/export-analyzer/main.go b/tools/export-analyzer/main.go
index 9891e6e..ab1b675 100644
--- a/tools/export-analyzer/main.go
+++ b/tools/export-analyzer/main.go
@@ -73,7 +73,7 @@ func main() {
 		log.Printf("Export signature file contents:\n%v", string(prettyJSON))
 	}
 
-	keyExport, err := export.UnmarshalExportFile(blob)
+	keyExport, _, err := export.UnmarshalExportFile(blob)
 	if err != nil {
 		log.Fatalf("error unmarshaling export file: %v", err)
 	}


Test patch:
diff --git a/internal/export/exportfile_test.go b/internal/export/exportfile_test.go
index 1dfbcfa..9cfdd62 100644
--- a/internal/export/exportfile_test.go
+++ b/internal/export/exportfile_test.go
@@ -16,6 +16,7 @@ package export
 
 import (
 	"crypto"
+	"encoding/base64"
 	"io"
 	"testing"
 	"time"
@@ -101,11 +102,16 @@ func TestMarshalUnmarshalExportFile(t *testing.T) {
 		t.Fatalf("Can't marshal export file, %v", err)
 	}
 
-	got, err := UnmarshalExportFile(blob)
+	got, digest, err := UnmarshalExportFile(blob)
 	if err != nil {
 		t.Fatalf("Unmarshal failed: %v", err)
 	}
 
+	wantDigest := "jN+W9DnqfXx5hp+6LaI8JuilsFWoiyF8DE/73OGZMJM="
+	if b64digest := base64.StdEncoding.EncodeToString(digest); b64digest != wantDigest {
+		t.Errorf("wrong message digest want: %v, got: %v", wantDigest, b64digest)
+	}
+
 	infos := []*export.SignatureInfo{
 		{
 			VerificationKeyVersion: proto.String("1"),
diff --git a/internal/exportimport/database/exportimport_test.go b/internal/exportimport/database/exportimport_test.go
new file mode 100644
index 0000000..a3842be
--- /dev/null
+++ b/internal/exportimport/database/exportimport_test.go
@@ -0,0 +1,249 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package database
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/x509"
+	"encoding/pem"
+	"testing"
+	"time"
+
+	"github.com/google/exposure-notifications-server/internal/database"
+	"github.com/google/exposure-notifications-server/internal/exportimport/model"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+)
+
+var (
+	approxTime = cmp.Options{cmpopts.EquateApproxTime(time.Second)}
+)
+
+func TestAddGetUpdateExportConfig(t *testing.T) {
+	t.Parallel()
+
+	testDB := database.NewTestDatabase(t)
+	ctx := context.Background()
+	exportImportDB := New(testDB)
+
+	fromTime := time.Now().UTC().Add(-1 * time.Second)
+	want := []*model.ExportImport{
+		{
+			IndexFile:  "https://mysever/exports/index.txt",
+			ExportRoot: "https://myserver/",
+			Region:     "US",
+			From:       fromTime,
+			Thru:       nil,
+		},
+	}
+	for _, w := range want {
+		if err := exportImportDB.AddConfig(ctx, w); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	got, err := exportImportDB.ActiveConfigs(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if diff := cmp.Diff(want, got, approxTime); diff != "" {
+		t.Errorf("mismatch (-want, +got):\n%s", diff)
+	}
+}
+
+func TestAddImportFiles(t *testing.T) {
+	t.Parallel()
+
+	testDB := database.NewTestDatabase(t)
+	ctx := context.Background()
+	exportImportDB := New(testDB)
+
+	now := time.Now().UTC()
+	config := model.ExportImport{
+		IndexFile:  "https://mysever/exports/index.txt",
+		ExportRoot: "https://myserver/",
+		Region:     "US",
+		From:       now,
+		Thru:       nil,
+	}
+	if err := exportImportDB.AddConfig(ctx, &config); err != nil {
+		t.Fatal(err)
+	}
+
+	filenames := []string{"a.zip", "b.zip", "c.zip"}
+
+	if n, err := exportImportDB.CreateFiles(ctx, &config, filenames); err != nil {
+		t.Fatal(err)
+	} else if n != len(filenames) {
+		t.Fatalf("incorrect number of files, want: %v, got: %v", len(filenames), n)
+	}
+
+	lockDuration := 15 * time.Minute
+	got, err := exportImportDB.GetOpenImportFiles(ctx, lockDuration, &config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var want []*model.ImportFile
+	for _, fname := range filenames {
+		want = append(want,
+			&model.ImportFile{
+				ExportImportID: config.ID,
+				ZipFilename:    fname,
+				Status:         model.ImportFileOpen,
+			})
+	}
+
+	opts := cmp.Options{cmpopts.IgnoreFields(model.ImportFile{}, "ID", "DiscoveredAt")}
+	if diff := cmp.Diff(want, got, opts); diff != "" {
+		t.Errorf("mismatch (-want, +got):\n%s", diff)
+	}
+}
+
+func TestLeaseAndCompleteImportFile(t *testing.T) {
+	t.Parallel()
+
+	testDB := database.NewTestDatabase(t)
+	ctx := context.Background()
+	exportImportDB := New(testDB)
+
+	lockDuration := 2 * time.Second
+	now := time.Now().UTC()
+	config := model.ExportImport{
+		IndexFile:  "https://mysever/exports/index.txt",
+		ExportRoot: "https://myserver/",
+		Region:     "US",
+		From:       now,
+		Thru:       nil,
+	}
+	if err := exportImportDB.AddConfig(ctx, &config); err != nil {
+		t.Fatal(err)
+	}
+
+	filenames := []string{"a.zip"}
+
+	if n, err := exportImportDB.CreateFiles(ctx, &config, filenames); err != nil {
+		t.Fatal(err)
+	} else if n != len(filenames) {
+		t.Fatalf("incorrect number of files, want: %v, got: %v", len(filenames), n)
+	}
+
+	openFiles, err := exportImportDB.GetOpenImportFiles(ctx, lockDuration, &config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if l := len(openFiles); l != 1 {
+		t.Fatalf("did't get expected files, want 1: got: %v", l)
+	}
+
+	testFile := openFiles[0]
+	if err := exportImportDB.LeaseImportFile(ctx, lockDuration, testFile); err != nil {
+		t.Fatalf("error locking file: %v", err)
+	}
+
+	if err := exportImportDB.LeaseImportFile(ctx, lockDuration, testFile); err == nil {
+		t.Fatalf("no error trying to lock already locked file")
+	}
+
+	time.Sleep(2 * lockDuration)
+	if err := exportImportDB.LeaseImportFile(ctx, lockDuration, testFile); err != nil {
+		t.Fatalf("unable to lock file where lock has expired: %v", err)
+	}
+
+	if err := exportImportDB.CompleteImportFile(ctx, testFile); err != nil {
+		t.Fatalf("unable to complete import file: %v", err)
+	}
+
+	openFiles, err = exportImportDB.GetOpenImportFiles(ctx, lockDuration, &config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if l := len(openFiles); l != 0 {
+		t.Fatalf("wrong number of open files, want: 0, got: %v", l)
+	}
+}
+
+func TestImportFilePublicKey(t *testing.T) {
+	t.Parallel()
+
+	testDB := database.NewTestDatabase(t)
+	ctx := context.Background()
+	exportImportDB := New(testDB)
+
+	now := time.Now().UTC()
+	config := model.ExportImport{
+		IndexFile:  "https://mysever/exports/index.txt",
+		ExportRoot: "https://myserver/",
+		Region:     "US",
+		From:       now,
+		Thru:       nil,
+	}
+	if err := exportImportDB.AddConfig(ctx, &config); err != nil {
+		t.Fatal(err)
+	}
+
+	// Generate test ECDSA key pair.
+	privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	publicKey := privateKey.Public()
+	// Get the PEM for the public key.
+	x509EncodedPub, err := x509.MarshalPKIXPublicKey(publicKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pemEncodedPub := pem.EncodeToMemory(&pem.Block{Type: "PUBLIC KEY", Bytes: x509EncodedPub})
+	pemPublicKey := string(pemEncodedPub)
+
+	// Create import file public key.
+	want := model.ImportFilePublicKey{
+		ExportImportID: config.ID,
+		KeyID:          "ghost",
+		KeyVersion:     "v1",
+		PublicKeyPEM:   pemPublicKey,
+		From:           time.Now().UTC().Add(-1 * time.Hour),
+		Thru:           nil,
+	}
+
+	if err := exportImportDB.AddImportFilePublicKey(ctx, &want); err != nil {
+		t.Fatalf("error adding public key: %v", err)
+	}
+
+	got, err := exportImportDB.AllowedKeys(ctx, &config)
+	if err != nil {
+		t.Fatalf("error reading public keys: %v", err)
+	}
+
+	if diff := cmp.Diff(&want, got[0], approxTime); diff != "" {
+		t.Errorf("mismatch (-want, +got):\n%s", diff)
+	}
+
+	if err := exportImportDB.ExpireImportFilePublicKey(ctx, &want); err != nil {
+		t.Fatalf("failed to expire key: %v", err)
+	}
+
+	got, err = exportImportDB.AllowedKeys(ctx, &config)
+	if err != nil {
+		t.Fatalf("error reading public keys: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("want 0 public keys, got: %v", got)
+	}
+}
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index e0f14b7..c5946ef 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -163,7 +163,7 @@ func TestIntegration(t *testing.T) {
 				}
 
 				// Process contents as an export
-				key, err := exportapi.UnmarshalExportFile(data)
+				key, _, err := exportapi.UnmarshalExportFile(data)
 				if err != nil {
 					return fmt.Errorf("failed to extract export data: %w", err)
 				}
diff --git a/internal/performance/export_test.go b/internal/performance/export_test.go
index d29acb7..174ac1d 100644
--- a/internal/performance/export_test.go
+++ b/internal/performance/export_test.go
@@ -198,7 +198,7 @@ func TestExport(t *testing.T) {
 			}
 
 			// Process contents as an export
-			key, err := exportapi.UnmarshalExportFile(data)
+			key, _, err := exportapi.UnmarshalExportFile(data)
 			if err != nil {
 				return fmt.Errorf("failed to extract export data: %v", err)
 			}
diff --git a/internal/publish/model/exposure_model_test.go b/internal/publish/model/exposure_model_test.go
index 73b86a8..90672b0 100644
--- a/internal/publish/model/exposure_model_test.go
+++ b/internal/publish/model/exposure_model_test.go
@@ -25,9 +25,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/exposure-notifications-server/internal/pb/export"
 	"github.com/google/exposure-notifications-server/internal/verification"
 	verifyapi "github.com/google/exposure-notifications-server/pkg/api/v1"
 	"github.com/google/exposure-notifications-server/pkg/base64util"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -1287,3 +1289,121 @@ func TestExposureReview(t *testing.T) {
 		})
 	}
 }
+
+func TestExposureFromExportFile(t *testing.T) {
+	validTEK := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}
+	invalidTEK := []byte{0, 1}
+
+	validInterval := int32(IntervalNumber(time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour)))
+
+	maxSymptomOnsetDays := int32(14)
+	cases := []struct {
+		name      string
+		key       *export.TemporaryExposureKey
+		want      *Exposure
+		wantError string
+	}{
+		{
+			name: "valid_key",
+			key: &export.TemporaryExposureKey{
+				KeyData:                    validTEK,
+				TransmissionRiskLevel:      proto.Int32(verifyapi.TransmissionRiskConfirmedStandard),
+				RollingStartIntervalNumber: proto.Int32(validInterval),
+				RollingPeriod:              proto.Int32(verifyapi.MaxIntervalCount),
+				ReportType:                 export.TemporaryExposureKey_CONFIRMED_TEST.Enum(),
+				DaysSinceOnsetOfSymptoms:   proto.Int32(2),
+			},
+			want: &Exposure{
+				ExposureKey:           validTEK,
+				TransmissionRisk:      verifyapi.TransmissionRiskConfirmedStandard,
+				IntervalNumber:        validInterval,
+				IntervalCount:         verifyapi.MaxIntervalCount,
+				LocalProvenance:       false,
+				ReportType:            verifyapi.ReportTypeConfirmed,
+				DaysSinceSymptomOnset: proto.Int32(2),
+			},
+		},
+		{
+			name: "invalid_key_length",
+			key: &export.TemporaryExposureKey{
+				KeyData:                    invalidTEK,
+				TransmissionRiskLevel:      proto.Int32(verifyapi.TransmissionRiskConfirmedStandard),
+				RollingStartIntervalNumber: proto.Int32(validInterval),
+				RollingPeriod:              proto.Int32(verifyapi.MaxIntervalCount),
+				ReportType:                 export.TemporaryExposureKey_CONFIRMED_TEST.Enum(),
+				DaysSinceOnsetOfSymptoms:   proto.Int32(2),
+			},
+			wantError: "invalid key length",
+		},
+		{
+			name: "trisk_too_low",
+			key: &export.TemporaryExposureKey{
+				KeyData:                    validTEK,
+				TransmissionRiskLevel:      proto.Int32(-1),
+				RollingStartIntervalNumber: proto.Int32(validInterval),
+				RollingPeriod:              proto.Int32(verifyapi.MaxIntervalCount),
+				ReportType:                 export.TemporaryExposureKey_CONFIRMED_TEST.Enum(),
+				DaysSinceOnsetOfSymptoms:   proto.Int32(2),
+			},
+			wantError: "transmission risk too low",
+		},
+		{
+			name: "trisk_too_high",
+			key: &export.TemporaryExposureKey{
+				KeyData:                    validTEK,
+				TransmissionRiskLevel:      proto.Int32(verifyapi.MaxTransmissionRisk + 1),
+				RollingStartIntervalNumber: proto.Int32(validInterval),
+				RollingPeriod:              proto.Int32(verifyapi.MaxIntervalCount),
+				ReportType:                 export.TemporaryExposureKey_CONFIRMED_TEST.Enum(),
+				DaysSinceOnsetOfSymptoms:   proto.Int32(2),
+			},
+			wantError: "transmission risk too high",
+		},
+		{
+			name: "unsupported_report_type",
+			key: &export.TemporaryExposureKey{
+				KeyData:                    validTEK,
+				TransmissionRiskLevel:      proto.Int32(verifyapi.MaxTransmissionRisk),
+				RollingStartIntervalNumber: proto.Int32(validInterval),
+				RollingPeriod:              proto.Int32(verifyapi.MaxIntervalCount),
+				ReportType:                 export.TemporaryExposureKey_RECURSIVE.Enum(),
+				DaysSinceOnsetOfSymptoms:   proto.Int32(2),
+			},
+			wantError: "unsupported report type",
+		},
+		{
+			name: "invalid_symptom_onset",
+			key: &export.TemporaryExposureKey{
+				KeyData:                    validTEK,
+				TransmissionRiskLevel:      proto.Int32(verifyapi.TransmissionRiskConfirmedStandard),
+				RollingStartIntervalNumber: proto.Int32(validInterval),
+				RollingPeriod:              proto.Int32(verifyapi.MaxIntervalCount),
+				ReportType:                 export.TemporaryExposureKey_CONFIRMED_TEST.Enum(),
+				DaysSinceOnsetOfSymptoms:   proto.Int32(56),
+			},
+			wantError: "days since onset of symptoms is out of range",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := FromExportKey(tc.key, maxSymptomOnsetDays)
+
+			if err != nil && tc.wantError == "" {
+				t.Fatalf("unexpected error: %v", err)
+			} else if err == nil && tc.wantError != "" {
+				t.Fatalf("missing expected error: %q", tc.wantError)
+			} else if err != nil && tc.wantError != "" && !strings.Contains(err.Error(), tc.wantError) {
+				t.Fatalf("wrong error, want: %q, got: %v", tc.wantError, err)
+			}
+
+			if tc.wantError != "" {
+				return
+			}
+
+			if diff := cmp.Diff(tc.want, got, cmpopts.IgnoreUnexported(Exposure{})); diff != "" {
+				t.Errorf("mismatch (-want, +got):\n%s", diff)
+			}
+		})
+	}
+}