Simplify the bridgedescs module. (10685242) · Commits · Legacy / gitolite / user / iwakeh / Collector

src/main/java/org/torproject/metrics/collector/bridgedescs/BridgeDescriptorParser.java

deleted100644 → 0

+0 −55

Original line number	Diff line number	Diff line
		/* Copyright 2010--2020 The Tor Project
		* See LICENSE for licensing information */

		package org.torproject.metrics.collector.bridgedescs;

		import org.slf4j.Logger;
		import org.slf4j.LoggerFactory;

		import java.io.BufferedReader;
		import java.io.IOException;
		import java.io.StringReader;
		import java.nio.charset.StandardCharsets;

		public class BridgeDescriptorParser {

		private SanitizedBridgesWriter sbw;

		private static final Logger logger = LoggerFactory.getLogger(
		BridgeDescriptorParser.class);

		/** Initializes a new bridge descriptor parser and links it to a
		* sanitized bridges writer to sanitize and store bridge descriptors. */
		public BridgeDescriptorParser(SanitizedBridgesWriter sbw) {
		if (null == sbw) {
		throw new IllegalArgumentException("SanitizedBridgesWriter has to be "
		+ "provided, but was null.");
		}
		this.sbw = sbw;
		}

		/** Parses the first line of the given descriptor data to determine the
		* descriptor type and passes it to the sanitized bridges writer. */
		public void parse(byte[] allData, String dateTime,
		String authorityFingerprint) {
		try {
		BufferedReader br = new BufferedReader(new StringReader(
		new String(allData, StandardCharsets.US_ASCII)));
		String line = br.readLine();
		if (line == null) {
		return;
		}
		if (line.startsWith("router ")) {
		this.sbw.sanitizeAndStoreServerDescriptor(allData);
		} else if (line.startsWith("extra-info ")) {
		this.sbw.sanitizeAndStoreExtraInfoDescriptor(allData);
		} else {
		this.sbw.sanitizeAndStoreNetworkStatus(allData, dateTime,
		authorityFingerprint);
		}
		} catch (IOException e) {
		logger.warn("Could not parse or write bridge descriptor.", e);
		}
		}
		}

src/main/java/org/torproject/metrics/collector/bridgedescs/BridgeSnapshotReader.java

deleted100644 → 0

+0 −248

Original line number	Diff line number	Diff line
		/* Copyright 2010--2020 The Tor Project
		* See LICENSE for licensing information */

		package org.torproject.metrics.collector.bridgedescs;

		import org.apache.commons.codec.binary.Hex;
		import org.apache.commons.codec.digest.DigestUtils;
		import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
		import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
		import org.slf4j.Logger;
		import org.slf4j.LoggerFactory;

		import java.io.BufferedInputStream;
		import java.io.BufferedReader;
		import java.io.BufferedWriter;
		import java.io.ByteArrayOutputStream;
		import java.io.File;
		import java.io.FileInputStream;
		import java.io.FileReader;
		import java.io.FileWriter;
		import java.io.IOException;
		import java.io.StringReader;
		import java.nio.charset.StandardCharsets;
		import java.util.Collections;
		import java.util.HashSet;
		import java.util.Set;
		import java.util.SortedSet;
		import java.util.Stack;
		import java.util.TreeSet;

		public class BridgeSnapshotReader {

		private static final Logger logger = LoggerFactory.getLogger(
		BridgeSnapshotReader.class);

		/**
		* Reads the half-hourly snapshots of bridge descriptors from Bifroest.
		*/
		public BridgeSnapshotReader(BridgeDescriptorParser bdp,
		File bridgeDirectoriesDir, File statsDirectory) {

		if (bdp == null \|\| bridgeDirectoriesDir == null
		\|\| statsDirectory == null) {
		throw new IllegalArgumentException();
		}

		SortedSet<String> parsed = new TreeSet<>();
		File pbdFile = new File(statsDirectory, "parsed-bridge-directories");
		boolean modified = false;
		if (bridgeDirectoriesDir.exists()) {
		if (pbdFile.exists()) {
		logger.debug("Reading file {}...", pbdFile.getAbsolutePath());
		try {
		BufferedReader br = new BufferedReader(new FileReader(pbdFile));
		String line;
		while ((line = br.readLine()) != null) {
		parsed.add(line);
		}
		br.close();
		logger.debug("Finished reading file {}.", pbdFile.getAbsolutePath());
		} catch (IOException e) {
		logger.warn("Failed reading file {}!", pbdFile.getAbsolutePath(), e);
		return;
		}
		}
		logger.debug("Importing files in directory {}/...", bridgeDirectoriesDir);
		Set<String> descriptorImportHistory = new HashSet<>();
		int parsedFiles = 0;
		int skippedFiles = 0;
		int parsedStatuses = 0;
		int parsedServerDescriptors = 0;
		int skippedServerDescriptors = 0;
		int parsedExtraInfoDescriptors = 0;
		int skippedExtraInfoDescriptors = 0;
		Stack<File> filesInInputDir = new Stack<>();
		filesInInputDir.add(bridgeDirectoriesDir);
		while (!filesInInputDir.isEmpty()) {
		File pop = filesInInputDir.pop();
		if (pop.isDirectory()) {
		Collections.addAll(filesInInputDir, pop.listFiles());
		} else if (!parsed.contains(pop.getName())) {
		try {
		FileInputStream in = new FileInputStream(pop);
		if (in.available() > 0) {
		TarArchiveInputStream tais;
		if (pop.getName().endsWith(".tar.gz")) {
		GzipCompressorInputStream gcis =
		new GzipCompressorInputStream(in);
		tais = new TarArchiveInputStream(gcis);
		} else if (pop.getName().endsWith(".tar")) {
		tais = new TarArchiveInputStream(in);
		} else {
		continue;
		}
		BufferedInputStream bis = new BufferedInputStream(tais);
		String fn = pop.getName();
		String[] fnParts = fn.split("-");
		if (fnParts.length != 5) {
		logger.warn("Invalid bridge descriptor tarball file name: {}. "
		+ "Skipping.", fn);
		continue;
		}
		String authorityPart = String.format("%s-%s-", fnParts[0],
		fnParts[1]);
		String datePart = String.format("%s-%s-%s", fnParts[2],
		fnParts[3], fnParts[4]);
		String authorityFingerprint;
		switch (authorityPart) {
		case "from-tonga-":
		authorityFingerprint =
		"4A0CCD2DDC7995083D73F5D667100C8A5831F16D";
		break;
		case "from-bifroest-":
		authorityFingerprint =
		"1D8F3A91C37C5D1C4C19B1AD1D0CFBE8BF72D8E1";
		break;
		case "from-serge-":
		authorityFingerprint =
		"BA44A889E64B93FAA2B114E02C2A279A8555C533";
		break;
		default:
		logger.warn("Did not recognize the bridge authority that "
		+ "generated {}. Skipping.", fn);
		continue;
		}
		String dateTime = datePart.substring(0, 10) + " "
		+ datePart.substring(11, 13) + ":"
		+ datePart.substring(13, 15) + ":"
		+ datePart.substring(15, 17);
		while ((tais.getNextTarEntry()) != null) {
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		int len;
		byte[] data = new byte[1024];
		while ((len = bis.read(data, 0, 1024)) >= 0) {
		baos.write(data, 0, len);
		}
		byte[] allData = baos.toByteArray();
		if (allData.length == 0) {
		continue;
		}
		String fileDigest = Hex.encodeHexString(DigestUtils.sha1(
		allData));
		String ascii = new String(allData, StandardCharsets.US_ASCII);
		BufferedReader br3 = new BufferedReader(new StringReader(
		ascii));
		String firstLine;
		do {
		firstLine = br3.readLine();
		} while (firstLine != null && firstLine.startsWith("@"));
		if (firstLine == null) {
		continue;
		}
		if (firstLine.startsWith("published ")
		\|\| firstLine.startsWith("flag-thresholds ")
		\|\| firstLine.startsWith("r ")) {
		bdp.parse(allData, dateTime, authorityFingerprint);
		parsedStatuses++;
		} else if (descriptorImportHistory.contains(fileDigest)) {
		/* Skip server descriptors or extra-info descriptors if
		* we parsed them before. */
		skippedFiles++;
		continue;
		} else {
		int start;
		int sig;
		int end = -1;
		String startToken = firstLine.startsWith("router ")
		? "router " : "extra-info ";
		String sigToken = "\nrouter-signature\n";
		String endToken = "\n-----END SIGNATURE-----\n";
		while (end < ascii.length()) {
		start = ascii.indexOf(startToken, end);
		if (start < 0) {
		break;
		}
		sig = ascii.indexOf(sigToken, start);
		if (sig < 0) {
		break;
		}
		sig += sigToken.length();
		end = ascii.indexOf(endToken, sig);
		if (end < 0) {
		break;
		}
		end += endToken.length();
		byte[] descBytes = new byte[end - start];
		System.arraycopy(allData, start, descBytes, 0,
		end - start);
		String descriptorDigest = Hex.encodeHexString(
		DigestUtils.sha1(descBytes));
		if (!descriptorImportHistory.contains(
		descriptorDigest)) {
		bdp.parse(descBytes, dateTime, authorityFingerprint);
		descriptorImportHistory.add(descriptorDigest);
		if (firstLine.startsWith("router ")) {
		parsedServerDescriptors++;
		} else {
		parsedExtraInfoDescriptors++;
		}
		} else {
		if (firstLine.startsWith("router ")) {
		skippedServerDescriptors++;
		} else {
		skippedExtraInfoDescriptors++;
		}
		}
		}
		}
		descriptorImportHistory.add(fileDigest);
		parsedFiles++;
		}
		bis.close();
		}
		in.close();

		/* Let's give some memory back, or we'll run out of it. */
		System.gc();

		parsed.add(pop.getName());
		modified = true;
		} catch (IOException e) {
		logger.warn("Could not parse bridge snapshot {}!", pop.getName(),
		e);
		}
		}
		}
		logger.debug("Finished importing files in directory {}/. In total, we "
		+ "parsed {} files (skipped {}) containing {} statuses, {} server "
		+ "descriptors (skipped {}), and {} extra-info descriptors (skipped "
		+ "{}).", bridgeDirectoriesDir, parsedFiles, skippedFiles,
		parsedStatuses, parsedServerDescriptors, skippedServerDescriptors,
		parsedExtraInfoDescriptors, skippedExtraInfoDescriptors);
		if (!parsed.isEmpty() && modified) {
		logger.debug("Writing file {}...", pbdFile.getAbsolutePath());
		pbdFile.getParentFile().mkdirs();
		try (BufferedWriter bw = new BufferedWriter(new FileWriter(pbdFile))) {
		for (String f : parsed) {
		bw.append(f).append("\n");
		}
		logger.debug("Finished writing file {}.", pbdFile.getAbsolutePath());
		} catch (IOException e) {
		logger.warn("Failed writing file {}!", pbdFile.getAbsolutePath(), e);
		}
		}
		}
		}
		}

src/main/java/org/torproject/metrics/collector/bridgedescs/SanitizedBridgesWriter.java

+223 −5

Original line number	Diff line number	Diff line
		@@ -19,12 +19,17 @@ import org.apache.commons.codec.DecoderException;
		import org.apache.commons.codec.binary.Base64;
		import org.apache.commons.codec.binary.Hex;
		import org.apache.commons.codec.digest.DigestUtils;
		import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
		import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
		import org.slf4j.Logger;
		import org.slf4j.LoggerFactory;

		import java.io.BufferedInputStream;
		import java.io.BufferedReader;
		import java.io.BufferedWriter;
		import java.io.ByteArrayOutputStream;
		import java.io.File;
		import java.io.FileInputStream;
		import java.io.FileReader;
		import java.io.FileWriter;
		import java.io.IOException;
		@@ -40,11 +45,17 @@ import java.time.LocalDateTime;
		import java.time.format.DateTimeFormatter;
		import java.time.temporal.ChronoUnit;
		import java.util.ArrayList;
		import java.util.Collections;
		import java.util.HashMap;
		import java.util.HashSet;
		import java.util.List;
		import java.util.Map;
		import java.util.Set;
		import java.util.SortedMap;
		import java.util.SortedSet;
		import java.util.Stack;
		import java.util.TreeMap;
		import java.util.TreeSet;

		/**
		* <p>Sanitizes bridge descriptors, i.e., removes all possibly sensitive
		@@ -217,12 +228,8 @@ public class SanitizedBridgesWriter extends CollecTorMain {
		logger.info("Using cut-off datetime '{}' for secrets.",
		this.bridgeSanitizingCutOffTimestamp);

		// Prepare bridge descriptor parser
		BridgeDescriptorParser bdp = new BridgeDescriptorParser(this);

		// Import bridge descriptors
		new BridgeSnapshotReader(bdp, this.bridgeDirectoriesDirectory,
		statsDirectory);
		this.readBridgeSnapshots(this.bridgeDirectoriesDirectory, statsDirectory);

		// Finish writing sanitized bridge descriptors to disk
		this.finishWriting();
		@@ -232,6 +239,217 @@ public class SanitizedBridgesWriter extends CollecTorMain {
		this.cleanUpDirectories();
		}

		private void readBridgeSnapshots(File bridgeDirectoriesDir,
		File statsDirectory) {

		if (bridgeDirectoriesDir == null
		\|\| statsDirectory == null) {
		throw new IllegalArgumentException();
		}

		SortedSet<String> parsed = new TreeSet<>();
		File pbdFile = new File(statsDirectory, "parsed-bridge-directories");
		boolean modified = false;
		if (bridgeDirectoriesDir.exists()) {
		if (pbdFile.exists()) {
		logger.debug("Reading file {}...", pbdFile.getAbsolutePath());
		try {
		BufferedReader br = new BufferedReader(new FileReader(pbdFile));
		String line;
		while ((line = br.readLine()) != null) {
		parsed.add(line);
		}
		br.close();
		logger.debug("Finished reading file {}.", pbdFile.getAbsolutePath());
		} catch (IOException e) {
		logger.warn("Failed reading file {}!", pbdFile.getAbsolutePath(), e);
		return;
		}
		}
		logger.debug("Importing files in directory {}/...", bridgeDirectoriesDir);
		Set<String> descriptorImportHistory = new HashSet<>();
		int parsedFiles = 0;
		int skippedFiles = 0;
		int parsedStatuses = 0;
		int parsedServerDescriptors = 0;
		int skippedServerDescriptors = 0;
		int parsedExtraInfoDescriptors = 0;
		int skippedExtraInfoDescriptors = 0;
		Stack<File> filesInInputDir = new Stack<>();
		filesInInputDir.add(bridgeDirectoriesDir);
		while (!filesInInputDir.isEmpty()) {
		File pop = filesInInputDir.pop();
		if (pop.isDirectory()) {
		Collections.addAll(filesInInputDir, pop.listFiles());
		} else if (!parsed.contains(pop.getName())) {
		try {
		FileInputStream in = new FileInputStream(pop);
		if (in.available() > 0) {
		TarArchiveInputStream tais;
		if (pop.getName().endsWith(".tar.gz")) {
		GzipCompressorInputStream gcis =
		new GzipCompressorInputStream(in);
		tais = new TarArchiveInputStream(gcis);
		} else if (pop.getName().endsWith(".tar")) {
		tais = new TarArchiveInputStream(in);
		} else {
		continue;
		}
		BufferedInputStream bis = new BufferedInputStream(tais);
		String fn = pop.getName();
		String[] fnParts = fn.split("-");
		if (fnParts.length != 5) {
		logger.warn("Invalid bridge descriptor tarball file name: {}. "
		+ "Skipping.", fn);
		continue;
		}
		String authorityPart = String.format("%s-%s-", fnParts[0],
		fnParts[1]);
		String datePart = String.format("%s-%s-%s", fnParts[2],
		fnParts[3], fnParts[4]);
		String authorityFingerprint;
		switch (authorityPart) {
		case "from-tonga-":
		authorityFingerprint =
		"4A0CCD2DDC7995083D73F5D667100C8A5831F16D";
		break;
		case "from-bifroest-":
		authorityFingerprint =
		"1D8F3A91C37C5D1C4C19B1AD1D0CFBE8BF72D8E1";
		break;
		case "from-serge-":
		authorityFingerprint =
		"BA44A889E64B93FAA2B114E02C2A279A8555C533";
		break;
		default:
		logger.warn("Did not recognize the bridge authority that "
		+ "generated {}. Skipping.", fn);
		continue;
		}
		String dateTime = datePart.substring(0, 10) + " "
		+ datePart.substring(11, 13) + ":"
		+ datePart.substring(13, 15) + ":"
		+ datePart.substring(15, 17);
		while ((tais.getNextTarEntry()) != null) {
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		int len;
		byte[] data = new byte[1024];
		while ((len = bis.read(data, 0, 1024)) >= 0) {
		baos.write(data, 0, len);
		}
		byte[] allData = baos.toByteArray();
		if (allData.length == 0) {
		continue;
		}
		String fileDigest = Hex.encodeHexString(DigestUtils.sha1(
		allData));
		String ascii = new String(allData, StandardCharsets.US_ASCII);
		BufferedReader br3 = new BufferedReader(new StringReader(
		ascii));
		String firstLine;
		do {
		firstLine = br3.readLine();
		} while (firstLine != null && firstLine.startsWith("@"));
		if (firstLine == null) {
		continue;
		}
		if (firstLine.startsWith("published ")
		\|\| firstLine.startsWith("flag-thresholds ")
		\|\| firstLine.startsWith("r ")) {
		this.sanitizeAndStoreNetworkStatus(allData, dateTime,
		authorityFingerprint);
		parsedStatuses++;
		} else if (descriptorImportHistory.contains(fileDigest)) {
		/* Skip server descriptors or extra-info descriptors if
		* we parsed them before. */
		skippedFiles++;
		continue;
		} else {
		int start;
		int sig;
		int end = -1;
		String startToken = firstLine.startsWith("router ")
		? "router " : "extra-info ";
		String sigToken = "\nrouter-signature\n";
		String endToken = "\n-----END SIGNATURE-----\n";
		while (end < ascii.length()) {
		start = ascii.indexOf(startToken, end);
		if (start < 0) {
		break;
		}
		sig = ascii.indexOf(sigToken, start);
		if (sig < 0) {
		break;
		}
		sig += sigToken.length();
		end = ascii.indexOf(endToken, sig);
		if (end < 0) {
		break;
		}
		end += endToken.length();
		byte[] descBytes = new byte[end - start];
		System.arraycopy(allData, start, descBytes, 0,
		end - start);
		String descriptorDigest = Hex.encodeHexString(
		DigestUtils.sha1(descBytes));
		if (!descriptorImportHistory.contains(
		descriptorDigest)) {
		descriptorImportHistory.add(descriptorDigest);
		if (firstLine.startsWith("router ")) {
		this.sanitizeAndStoreServerDescriptor(descBytes);
		parsedServerDescriptors++;
		} else {
		this.sanitizeAndStoreExtraInfoDescriptor(descBytes);
		parsedExtraInfoDescriptors++;
		}
		} else {
		if (firstLine.startsWith("router ")) {
		skippedServerDescriptors++;
		} else {
		skippedExtraInfoDescriptors++;
		}
		}
		}
		}
		descriptorImportHistory.add(fileDigest);
		parsedFiles++;
		}
		bis.close();
		}
		in.close();

		/* Let's give some memory back, or we'll run out of it. */
		System.gc();

		parsed.add(pop.getName());
		modified = true;
		} catch (IOException e) {
		logger.warn("Could not parse bridge snapshot {}!", pop.getName(),
		e);
		}
		}
		}
		logger.debug("Finished importing files in directory {}/. In total, we "
		+ "parsed {} files (skipped {}) containing {} statuses, {} server "
		+ "descriptors (skipped {}), and {} extra-info descriptors (skipped "
		+ "{}).", bridgeDirectoriesDir, parsedFiles, skippedFiles,
		parsedStatuses, parsedServerDescriptors, skippedServerDescriptors,
		parsedExtraInfoDescriptors, skippedExtraInfoDescriptors);
		if (!parsed.isEmpty() && modified) {
		logger.debug("Writing file {}...", pbdFile.getAbsolutePath());
		pbdFile.getParentFile().mkdirs();
		try (BufferedWriter bw = new BufferedWriter(new FileWriter(pbdFile))) {
		for (String f : parsed) {
		bw.append(f).append("\n");
		}
		logger.debug("Finished writing file {}.", pbdFile.getAbsolutePath());
		} catch (IOException e) {
		logger.warn("Failed writing file {}!", pbdFile.getAbsolutePath(), e);
		}
		}
		}
		}

		private String scrubOrAddress(String orAddress, byte[] fingerprintBytes,
		String published) throws IOException {
		if (!orAddress.contains(":")) {

src/test/java/org/torproject/metrics/collector/bridgedescs/BridgeDescriptorParserTest.java

deleted100644 → 0

+0 −43

Original line number	Diff line number	Diff line
		/* Copyright 2016--2020 The Tor Project
		* See LICENSE for licensing information */

		package org.torproject.metrics.collector.bridgedescs;

		import org.torproject.metrics.collector.conf.Configuration;

		import org.junit.Test;

		public class BridgeDescriptorParserTest {

		@Test(expected = IllegalArgumentException.class)
		public void testNullArgForConstructor() {
		new BridgeDescriptorParser(null);
		}

		@Test(expected = NullPointerException.class)
		public void testNullData() {
		BridgeDescriptorParser bdp = new BridgeDescriptorParser(
		new SanitizedBridgesWriter(new Configuration()));
		bdp.parse(null, "", "");
		}

		@Test
		/* Empty data is not passed down to the sanitized writer.
		* This test passes when there is no exception. */
		public void testDataEmpty() {
		BridgeDescriptorParser bdp = new BridgeDescriptorParser(
		new SanitizedBridgesWriter(new Configuration()));
		bdp.parse(new byte[]{}, null, null);
		}

		@Test(expected = NullPointerException.class)
		/* The SanitizedBridgesWriter wasn't initialized sufficiently.
		* Actually that should be corrected in SanitizedBridgesWriter
		* at some point, but that's a bigger rewrite. */
		public void testMinimalData() {
		BridgeDescriptorParser bdp = new BridgeDescriptorParser(
		new SanitizedBridgesWriter(new Configuration()));
		bdp.parse(new byte[]{0}, "2010-10-10 10:10:10", null);
		}

		}