This action is invoked by {@link RdeStagingAction} once it's created the files we need. The * date is calculated from {@link CursorType#RDE_UPLOAD}. * *
Once this action completes, it rolls the cursor forward a day and triggers * {@link RdeReportAction}. */ @Action(path = RdeUploadAction.PATH, method = POST) public final class RdeUploadAction implements Runnable, EscrowTask { static final String PATH = "/_dr/task/rdeUpload"; private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass(); @Inject Clock clock; @Inject GcsUtils gcsUtils; @Inject Ghostryde ghostryde; @Inject EscrowTaskRunner runner; @Inject JSch jsch; @Inject JSchSshSessionFactory jschSshSessionFactory; @Inject Response response; @Inject RydePgpCompressionOutputStreamFactory pgpCompressionFactory; @Inject RydePgpEncryptionOutputStreamFactory pgpEncryptionFactory; @Inject RydePgpFileOutputStreamFactory pgpFileFactory; @Inject RydePgpSigningOutputStreamFactory pgpSigningFactory; @Inject RydeTarOutputStreamFactory tarFactory; @Inject TaskEnqueuer taskEnqueuer; @Inject @Parameter(RequestParameters.PARAM_TLD) String tld; @Inject @Config("rdeBucket") String bucket; @Inject @Config("rdeInterval") Duration interval; @Inject @Config("rdeUploadLockTimeout") Duration timeout; @Inject @Config("rdeUploadSftpCooldown") Duration sftpCooldown; @Inject @Config("rdeUploadUrl") URI uploadUrl; @Inject @Key("rdeReceiverKey") PGPPublicKey receiverKey; @Inject @Key("rdeSigningKey") PGPKeyPair signingKey; @Inject @Key("rdeStagingDecryptionKey") PGPPrivateKey stagingDecryptionKey; @Inject @Named("rde-report") Queue reportQueue; @Inject RdeUploadAction() {} @Override public void run() { runner.lockRunAndRollForward(this, Registry.get(tld), timeout, CursorType.RDE_UPLOAD, interval); taskEnqueuer.enqueue( reportQueue, withUrl(RdeReportAction.PATH).param(RequestParameters.PARAM_TLD, tld)); } @Override public void runWithLock(DateTime watermark) throws Exception { DateTime stagingCursorTime = getCursorTimeOrStartOfTime( ofy().load().key(Cursor.createKey(CursorType.RDE_STAGING, Registry.get(tld))).now()); if (!stagingCursorTime.isAfter(watermark)) { logger.infofmt("tld=%s uploadCursor=%s stagingCursor=%s", tld, watermark, stagingCursorTime); throw new ServiceUnavailableException("Waiting for RdeStagingAction to complete"); } DateTime sftpCursorTime = getCursorTimeOrStartOfTime( ofy().load().key(Cursor.createKey(CursorType.RDE_UPLOAD_SFTP, Registry.get(tld))).now()); if (sftpCursorTime.plus(sftpCooldown).isAfter(clock.nowUtc())) { // Fail the task good and hard so it retries until the cooldown passes. logger.infofmt("tld=%s cursor=%s sftpCursor=%s", tld, watermark, sftpCursorTime); throw new ServiceUnavailableException("SFTP cooldown has not yet passed"); } int revision = RdeRevision.getNextRevision(tld, watermark, FULL) - 1; verify(revision >= 0, "RdeRevision was not set on generated deposit"); String name = RdeNamingUtils.makeRydeFilename(tld, watermark, FULL, 1, revision); GcsFilename xmlFilename = new GcsFilename(bucket, name + ".xml.ghostryde"); GcsFilename xmlLengthFilename = new GcsFilename(bucket, name + ".xml.length"); GcsFilename reportFilename = new GcsFilename(bucket, name + "-report.xml.ghostryde"); verifyFileExists(xmlFilename); verifyFileExists(xmlLengthFilename); verifyFileExists(reportFilename); upload(xmlFilename, readXmlLength(xmlLengthFilename), watermark, name); ofy().transact(new VoidWork() { @Override public void vrun() { Cursor cursor = Cursor.create( CursorType.RDE_UPLOAD_SFTP, ofy().getTransactionTime(), Registry.get(tld)); ofy().save().entity(cursor).now(); } }); response.setContentType(PLAIN_TEXT_UTF_8); response.setPayload(String.format("OK %s %s\n", tld, watermark)); } /** * Performs a blocking upload of a cloud storage XML file to escrow provider, converting * it to the RyDE format along the way by applying tar+compress+encrypt+sign, and saving the * created RyDE file on GCS for future reference. * *
This is done by layering a bunch of {@link java.io.FilterOutputStream FilterOutputStreams} * on top of each other in reverse order that turn XML bytes into a RyDE file while * simultaneously uploading it to the SFTP endpoint, and then using {@link ByteStreams#copy} to * blocking-copy bytes from the cloud storage {@code InputStream} to the RyDE/SFTP pipeline. * *
In psuedoshell, the whole process looks like the following: * *
{@code
* gcs read $xmlFile \ # Get GhostRyDE from cloud storage.
* | decrypt | decompress \ # Convert it to XML.
* | tar | file | compress | encrypt | sign /tmp/sig \ # Convert it to a RyDE file.
* | tee gs://bucket/$rydeFilename.ryde \ # Save a copy of the RyDE file to GCS.
* | sftp put $dstUrl/$rydeFilename.ryde \ # Upload to SFTP server.
* && sftp put $dstUrl/$rydeFilename.sig gs://bucket/$rydeFilename.sig # Save a copy of signature to GCS.
* }
*/
private void upload(
GcsFilename xmlFile, long xmlLength, DateTime watermark, String name) throws Exception {
logger.infofmt("Uploading %s to %s", xmlFile, uploadUrl);
try (InputStream gcsInput = gcsUtils.openInputStream(xmlFile);
Ghostryde.Decryptor decryptor = ghostryde.openDecryptor(gcsInput, stagingDecryptionKey);
Ghostryde.Decompressor decompressor = ghostryde.openDecompressor(decryptor);
Ghostryde.Input xmlInput = ghostryde.openInput(decompressor)) {
try (JSchSshSession session = jschSshSessionFactory.create(jsch, uploadUrl);
JSchSftpChannel ftpChan = session.openSftpChannel()) {
byte[] signature;
String rydeFilename = name + ".ryde";
GcsFilename rydeGcsFilename = new GcsFilename(bucket, rydeFilename);
try (OutputStream ftpOutput = ftpChan.get().put(rydeFilename, OVERWRITE);
OutputStream gcsOutput = gcsUtils.openOutputStream(rydeGcsFilename);
TeeOutputStream teeOutput = new TeeOutputStream(asList(ftpOutput, gcsOutput));
RydePgpSigningOutputStream signer = pgpSigningFactory.create(teeOutput, signingKey)) {
try (OutputStream encryptLayer = pgpEncryptionFactory.create(signer, receiverKey);
OutputStream kompressor = pgpCompressionFactory.create(encryptLayer);
OutputStream fileLayer = pgpFileFactory.create(kompressor, watermark, name + ".tar");
OutputStream tarLayer =
tarFactory.create(fileLayer, xmlLength, watermark, name + ".xml")) {
ByteStreams.copy(xmlInput, tarLayer);
}
signature = signer.getSignature();
logger.infofmt("uploaded %,d bytes: %s.ryde", signer.getBytesWritten(), name);
}
String sigFilename = name + ".sig";
gcsUtils.createFromBytes(new GcsFilename(bucket, sigFilename), signature);
ftpChan.get().put(new ByteArrayInputStream(signature), sigFilename);
logger.infofmt("uploaded %,d bytes: %s.sig", signature.length, name);
}
}
}
/** Reads the contents of a file from Cloud Storage that contains nothing but an integer. */
private long readXmlLength(GcsFilename xmlLengthFilename) throws IOException {
try (InputStream input = gcsUtils.openInputStream(xmlLengthFilename)) {
return Long.parseLong(new String(ByteStreams.toByteArray(input), UTF_8).trim());
}
}
private void verifyFileExists(GcsFilename filename) {
verify(gcsUtils.existsAndNotEmpty(filename), "Missing file: %s", filename);
}
}