diff --git a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java index b1ced657..55c97ae7 100644 --- a/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java +++ b/OpenICF-csvfile-connector/src/main/java/org/forgerock/openicf/csvfile/CSVFileConnector.java @@ -463,10 +463,13 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha final String[] header = getHeader(); File syncOrigin = null; + // the snapshots this call copies from the live file; deleted again if the sync fails + final List copies = new ArrayList(2); boolean changesProcessed = false; // start with a write-lock final ReentrantReadWriteLock rwLock = fileNameToLockMap.get(csvFilePath); + boolean downgraded = false; rwLock.writeLock().lock(); try { if (token != null && token.getValue() != null) { @@ -482,17 +485,26 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } catch (IOException e) { throw new ConnectorException("Unable to copy CSV file for sync operation", e); } + copies.add(syncOrigin); token = new SyncToken(timestamp); } scrubSyncFiles(); // downgrade to a read-lock rwLock.readLock().lock(); + downgraded = true; } finally { + if (!downgraded) { + // the read section below deletes the copies only once it is entered + for (File copy : copies) { + deleteRewriteFile(copy); + } + } rwLock.writeLock().unlock(); } // we now have a read-lock unless the above block threw an Exception + boolean completed = false; try { final MessageDigest digest = newFingerprintDigest(); // uid of every current row -> fingerprint of the first origin row with that uid, null while there is none @@ -562,10 +574,19 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha boolean reportDeletes = true; while ((entry = reader.read(originHeader, processors)) != null) { final String uid = (String) entry.get(config.getHeaderUid()); + if (StringUtil.isBlank(uid)) { + // a row without a uid was never reported, so there is no object to delete + log.warn("Skipping line {0} of {1}: it has no {2} value", + String.valueOf(reader.getLineNumber()), syncOrigin.toString(), config.getHeaderUid()); + continue; + } if (!index.containsKey(uid)) { if (reportDeletes) { SyncDelta delta = generateSyncDelta(SyncDeltaType.DELETE, newConnectorObject(entry), token); - if (handler.handle(delta)) { + final boolean more = handler.handle(delta); + // the handler may have stored the token of this delta, so its snapshot must stay + copies.clear(); + if (more) { changesProcessed = true; } else { // keep reading: the next pass needs the fingerprints of the remaining origin rows @@ -612,7 +633,10 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha continue; } SyncDelta delta = generateSyncDelta(type, newConnectorObject(entry), token); - if (!handler.handle(delta)) { + final boolean more = handler.handle(delta); + // the handler may have stored the token of this delta, so its snapshot must stay + copies.clear(); + if (!more) { break; } changesProcessed = true; @@ -643,6 +667,7 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha } catch (IOException e) { throw new ConnectorException("Unable to copy CSV file for sync operation", e); } + copies.add(syncOrigin); token = new SyncToken(timestamp); } } @@ -650,8 +675,16 @@ public void sync(ObjectClass objectClass, SyncToken token, SyncResultsHandler ha if (handler instanceof SyncTokenResultsHandler) { ((SyncTokenResultsHandler) handler).handleResult(token); } + completed = true; } finally { rwLock.readLock().unlock(); + if (!completed) { + // a kept copy could be handed out by getLatestSyncToken, + // and it would stop a retry from copying the unchanged live file again + for (File copy : copies) { + deleteSnapshot(rwLock, copy); + } + } } } @@ -999,6 +1032,18 @@ private void scrubSyncFiles() { } } + /** + * Deletes a snapshot under the write lock, so that no sync in this JVM is reading it meanwhile. + */ + private void deleteSnapshot(ReentrantReadWriteLock rwLock, File snapshot) { + rwLock.writeLock().lock(); + try { + deleteRewriteFile(snapshot); + } finally { + rwLock.writeLock().unlock(); + } + } + private void compareHeaders(String[] header2) { List hdr1 = Arrays.asList(getHeader()); List hdr2 = Arrays.asList(header2); diff --git a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java index 22822549..d38b495f 100644 --- a/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java +++ b/OpenICF-csvfile-connector/src/test/java/org/forgerock/openicf/csvfile/SyncOpDiffTest.java @@ -16,18 +16,29 @@ package org.forgerock.openicf.csvfile; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotEquals; +import static org.testng.Assert.assertTrue; import static org.testng.Assert.expectThrows; +import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; +import java.io.PrintStream; +import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; +import java.nio.file.FileSystems; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.nio.file.attribute.PosixFilePermission; +import java.nio.file.attribute.PosixFilePermissions; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; import java.util.stream.Stream; import org.identityconnectors.common.security.GuardedString; @@ -43,6 +54,7 @@ import org.identityconnectors.framework.common.objects.SyncToken; import org.identityconnectors.framework.spi.SyncTokenResultsHandler; import org.supercsv.exception.SuperCsvException; +import org.testng.SkipException; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; @@ -58,6 +70,10 @@ public class SyncOpDiffTest { private static final SyncToken SNAPSHOT_TOKEN = new SyncToken(1300000000000L); + // the second row of each uid differs from the first one, so even a fresh copy of this file reports two UPDATEs + private static final String[] DUPLICATE_UIDS = + { HEADER, "x,Jane,Doe,pw1", "x,Jane,Changed,pw1", "y,John,Roe,pw2", "y,John,Other,pw2" }; + private Path dir; private File csvFile; @@ -176,6 +192,163 @@ public void rowWithoutUidFailsSyncBeforeAnyDelta(String row) throws Exception { assertEquals(e.getMessage(), "Line 3 of " + csvFile + " has no uid value"); } + @Test(dataProvider = "rowsWithoutUid") + public void snapshotRowWithoutUidIsSkipped(String row) throws Exception { + givenSnapshot(HEADER, "u1,Ann,Roe,pw0", row, "u3,Bob,Poe,pw3"); + givenLive(HEADER, "u3,Bob,Poe,pw3", "u2,Jane,Doe,pw1"); + + assertEquals(sync(SNAPSHOT_TOKEN), Arrays.asList("DELETE u1 [Ann, Roe, pw0]", "CREATE u2 [Jane, Doe, pw1]")); + } + + @Test + public void skippedSnapshotRowIsLogged() throws Exception { + givenSnapshot(HEADER, "u1,Jane,Doe,pw1", ",John,Roe,pw2"); + givenLive(HEADER, "u1,Jane,Doe,pw1"); + + String output = captureStdOut(() -> sync(SNAPSHOT_TOKEN)); + + assertTrue(output.contains("Skipping line 3 of " + + dir.resolve(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue()) + ": it has no uid value"), output); + } + + @Test + public void failedSyncDeletesTheSnapshotItCopied() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1", ",John,Roe,pw2"); + + expectThrows(IllegalArgumentException.class, () -> sync(null)); + + assertEquals(snapshots(), Collections.emptyList()); + } + + @Test + public void failedSyncKeepsTheSnapshotOfItsToken() throws Exception { + givenSnapshot(HEADER, "u1,Jane,Doe,pw1"); + givenLive(HEADER, "u1,Jane,Doe,pw1", ",John,Roe,pw2"); + + expectThrows(IllegalArgumentException.class, () -> sync(SNAPSHOT_TOKEN)); + + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue())); + } + + @Test + public void successfulSyncKeepsTheSnapshotItCopied() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1"); + + sync(null); + + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + resultToken.getValue())); + } + + @Test + public void syncWhoseTokenIsNotDeliveredDeletesTheSnapshotItCopied() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1"); + + expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(null, () -> { })); + + assertEquals(snapshots(), Collections.emptyList()); + } + + @Test + public void syncWhoseTokenIsNotDeliveredDeletesTheSnapshotItCopiedAfterTheDiff() throws Exception { + givenSnapshot(HEADER, "u1,Jane,Doe,pw1"); + givenLive(HEADER, "u1,Jane,Doe,pw1", "u2,John,Roe,pw2"); + + expectThrows(IllegalStateException.class, () -> syncWithFailingHandleResult(SNAPSHOT_TOKEN, () -> { })); + + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + SNAPSHOT_TOKEN.getValue())); + } + + @Test + public void failedSyncKeepsTheSnapshotOfADeltaItHandled() throws Exception { + givenLive(DUPLICATE_UIDS); + + List handled = syncFailingAfter(1); + + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + handled.get(0).getValue())); + } + + @Test + public void failedSyncKeepsTheSnapshotOfADeleteItHandled() throws Exception { + givenLive(HEADER, "u1,Jane,Doe,pw1", "u2,John,Roe,pw2"); + connector(); + // relies on the header cached at init: the live rows are read with it, so every live uid is a password, + // while a fresh copy is read with its own header and reports two DELETEs + givenLive("password,lastName,uid,firstName", "pw1,Doe,u1,Jane", "pw2,Roe,u2,John"); + + List handled = syncFailingAfter(1); + + assertEquals(snapshots(), Collections.singletonList(csvFile.getName() + "." + handled.get(0).getValue())); + } + + @Test + public void syncWhoseFirstDeltaFailsDeletesTheSnapshotItCopied() throws Exception { + givenLive(DUPLICATE_UIDS); + + syncFailingAfter(0); + + assertEquals(snapshots(), Collections.emptyList()); + } + + @Test + public void snapshotThatCannotBeDeletedIsLogged() throws Exception { + if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { + throw new SkipException("A read-only directory needs POSIX permissions"); + } + final Set mode = Files.getPosixFilePermissions(dir); + final Set readOnly = PosixFilePermissions.fromString("r-xr-xr-x"); + setDirMode(readOnly); + final boolean writable = Files.isWritable(dir); + setDirMode(mode); + if (writable) { + throw new SkipException("A read-only directory does not stop this user from deleting files"); + } + givenLive(HEADER, "u1,Jane,Doe,pw1"); + final File copy = new File(dir.toFile(), csvFile.getName() + "." + csvFile.lastModified()); + + final String output; + try { + output = captureStdOut(() -> expectThrows(IllegalStateException.class, + () -> syncWithFailingHandleResult(null, () -> setDirMode(readOnly)))); + } finally { + setDirMode(mode); + } + + assertTrue(output.contains("Could not delete " + copy), output); + } + + @Test + public void syncThatCannotListItsSnapshotsDeletesTheSnapshotItCopied() throws Exception { + if (!FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) { + throw new SkipException("An unlistable directory needs POSIX permissions"); + } + final Set mode = Files.getPosixFilePermissions(dir); + // files can still be created in the directory, but not listed + final Set writeOnly = PosixFilePermissions.fromString("-wx------"); + setDirMode(writeOnly); + final boolean listable = dir.toFile().list() != null; + setDirMode(mode); + if (listable) { + throw new SkipException("A directory without read permission does not stop this user from listing it"); + } + givenLive(HEADER, "u1,Jane,Doe,pw1"); + connector(); + // only creating or deleting an entry in the directory moves its mtime off this value + final FileTime untouched = FileTime.fromMillis(0); + Files.setLastModifiedTime(dir, untouched); + + setDirMode(writeOnly); + try { + // the copy is made, then scrubSyncFiles fails while the sync still holds the write lock; + // how it fails is not the point here + expectThrows(RuntimeException.class, () -> sync(null)); + } finally { + setDirMode(mode); + } + + assertNotEquals(Files.getLastModifiedTime(dir), untouched, "The sync failed before it copied the CSV file"); + assertEquals(snapshots(), Collections.emptyList()); + } + @Test public void rowWithWrongColumnCountFailsSyncBeforeAnyDelta() throws Exception { givenSnapshot(HEADER, "u1,Jane,Doe,pw1"); @@ -298,6 +471,64 @@ private void givenLive(String... lines) throws IOException { Files.write(csvFile.toPath(), Arrays.asList(lines), StandardCharsets.UTF_8); } + // the connector logs through StdOutLogger, which reads System.out on every call + private static String captureStdOut(Runnable action) { + final PrintStream original = System.out; + final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + System.setOut(new PrintStream(buffer, true)); + try { + action.run(); + } finally { + System.setOut(original); + } + return buffer.toString(); + } + + private List snapshots() throws IOException { + try (Stream paths = Files.list(dir)) { + return paths.map(path -> path.getFileName().toString()) + .filter(name -> name.startsWith(csvFile.getName() + ".")) + .sorted() + .collect(Collectors.toList()); + } + } + + private void syncWithFailingHandleResult(final SyncToken token, final Runnable beforeFailing) { + connector().sync(ObjectClass.ACCOUNT, token, new SyncTokenResultsHandler() { + @Override + public boolean handle(SyncDelta delta) { + return true; + } + + @Override + public void handleResult(SyncToken syncToken) { + beforeFailing.run(); + throw new IllegalStateException("token not stored"); + } + }, null); + } + + // syncs from no token, handles the given number of deltas and fails on the next one + private List syncFailingAfter(final int count) { + final List handled = new ArrayList<>(); + expectThrows(IllegalStateException.class, () -> connector().sync(ObjectClass.ACCOUNT, null, delta -> { + if (handled.size() == count) { + throw new IllegalStateException("delta not stored"); + } + handled.add(delta.getToken()); + return true; + }, null)); + return handled; + } + + private void setDirMode(Set mode) { + try { + Files.setPosixFilePermissions(dir, mode); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + private CSVFileConnector connector() { if (connector == null) { CSVFileConfiguration config = new CSVFileConfiguration();