Github user cestella commented on a diff in the pull request:
https://github.com/apache/incubator-metron/pull/146#discussion_r66321208
--- Diff:
metron-platform/metron-data-management/src/main/java/org/apache/metron/dataloads/nonbulk/taxii/TaxiiHandler.java
---
@@ -61,346 +63,334 @@
import javax.xml.transform.TransformerFactory;
import javax.xml.transform.dom.DOMSource;
import javax.xml.transform.stream.StreamResult;
-import java.io.*;
+import java.io.IOException;
+import java.io.StringWriter;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.*;
public class TaxiiHandler extends TimerTask {
- private static final Logger LOG = Logger.getLogger(TaxiiHandler.class);
+ private static final Logger LOG = Logger.getLogger(TaxiiHandler.class);
- private static ThreadLocal<TaxiiXmlFactory> xmlFactory = new
ThreadLocal<TaxiiXmlFactory>() {
- @Override
- protected TaxiiXmlFactory initialValue() {
- return new TaxiiXmlFactory();
- }
- };
- private static ThreadLocal<ObjectFactory> messageFactory = new
ThreadLocal<ObjectFactory>() {
- @Override
- protected ObjectFactory initialValue() {
- return new ObjectFactory();
- }
- };
-
- private HttpClient taxiiClient;
- private URL endpoint;
- private Extractor extractor;
- private String hbaseTable;
- private String columnFamily;
- private Map<String, HTableInterface> connectionCache = new HashMap<>();
- private HttpClientContext context;
- private String collection;
- private String subscriptionId;
- private EnrichmentConverter converter = new EnrichmentConverter();
- private Date beginTime;
- private Configuration config;
- private boolean inProgress = false;
- private Set<String> allowedIndicatorTypes;
- public TaxiiHandler( TaxiiConnectionConfig connectionConfig
- , Extractor extractor
- , Configuration config
- ) throws Exception
- {
- LOG.info("Loading configuration: " + connectionConfig);
- this.allowedIndicatorTypes =
connectionConfig.getAllowedIndicatorTypes();
- this.extractor = extractor;
- this.collection = connectionConfig.getCollection();
- this.subscriptionId = connectionConfig.getSubscriptionId();
- hbaseTable = connectionConfig.getTable();
- columnFamily = connectionConfig.getColumnFamily();
- this.beginTime = connectionConfig.getBeginTime();
- this.config = config;
- initializeClient(connectionConfig);
- LOG.info("Configured, starting polling " + endpoint + " for " +
collection);
+ private static ThreadLocal<TaxiiXmlFactory> xmlFactory = new
ThreadLocal<TaxiiXmlFactory>() {
+ @Override
+ protected TaxiiXmlFactory initialValue() {
+ return new TaxiiXmlFactory();
}
+ };
+ private static ThreadLocal<ObjectFactory> messageFactory = new
ThreadLocal<ObjectFactory>() {
+ @Override
+ protected ObjectFactory initialValue() {
+ return new ObjectFactory();
+ }
+ };
- protected synchronized HTableInterface getTable(String table) throws
IOException {
- HTableInterface ret = connectionCache.get(table);
- if(ret == null) {
- ret = createHTable(table);
- connectionCache.put(table, ret);
- }
- return ret;
+ private HttpClient taxiiClient;
+ private URL endpoint;
+ private Extractor extractor;
+ private String hbaseTable;
+ private String columnFamily;
+ private Map<String, HTableInterface> connectionCache = new HashMap<>();
+ private HttpClientContext context;
+ private String collection;
+ private String subscriptionId;
+ private EnrichmentConverter converter = new EnrichmentConverter();
+ private Date beginTime;
+ private Configuration config;
+ private boolean inProgress = false;
+ private Set<String> allowedIndicatorTypes;
+ public TaxiiHandler( TaxiiConnectionConfig connectionConfig
+ , Extractor extractor
+ , Configuration config
+ ) throws Exception
+ {
+ LOG.info("Loading configuration: " + connectionConfig);
+ this.allowedIndicatorTypes =
connectionConfig.getAllowedIndicatorTypes();
+ this.extractor = extractor;
+ this.collection = connectionConfig.getCollection();
+ this.subscriptionId = connectionConfig.getSubscriptionId();
+ hbaseTable = connectionConfig.getTable();
+ columnFamily = connectionConfig.getColumnFamily();
+ this.beginTime = connectionConfig.getBeginTime();
+ this.config = config;
+ initializeClient(connectionConfig);
+ LOG.info("Configured, starting polling " + endpoint + " for " +
collection);
+ }
+
+ protected synchronized HTableInterface getTable(String table) throws
IOException {
+ HTableInterface ret = connectionCache.get(table);
+ if(ret == null) {
+ ret = createHTable(table);
+ connectionCache.put(table, ret);
}
+ return ret;
+ }
- protected synchronized HTableInterface createHTable(String tableInfo)
throws IOException {
- return new HTable(config, tableInfo);
+ protected synchronized HTableInterface createHTable(String tableInfo)
throws IOException {
+ return new HTable(config, tableInfo);
+ }
+ /**
+ * The action to be performed by this timer task.
+ */
+ @Override
+ public void run() {
+ if(inProgress) {
+ return;
}
- /**
- * The action to be performed by this timer task.
- */
- @Override
- public void run() {
- if(inProgress) {
- return;
- }
- Date ts = new Date();
- LOG.info("Polling..." + new SimpleDateFormat().format(ts));
+ Date ts = new Date();
+ LOG.info("Polling..." + new SimpleDateFormat().format(ts));
+ try {
+ inProgress = true;
+ // Prepare the message to send.
+ String sessionID = MessageHelper.generateMessageId();
+ PollRequest request = messageFactory.get().createPollRequest()
+ .withMessageId(sessionID)
+ .withCollectionName(collection);
+ if (subscriptionId != null) {
+ request = request.withSubscriptionID(subscriptionId);
+ } else {
+ request =
request.withPollParameters(messageFactory.get().createPollParametersType());
+ }
+ if (beginTime != null) {
+ Calendar gc = GregorianCalendar.getInstance();
+ gc.setTime(beginTime);
+ XMLGregorianCalendar gTime = null;
try {
- inProgress = true;
- // Prepare the message to send.
- String sessionID = MessageHelper.generateMessageId();
- PollRequest request = messageFactory.get().createPollRequest()
- .withMessageId(sessionID)
- .withCollectionName(collection);
- if (subscriptionId != null) {
- request = request.withSubscriptionID(subscriptionId);
- } else {
- request =
request.withPollParameters(messageFactory.get().createPollParametersType());
- }
- if (beginTime != null) {
- Calendar gc = GregorianCalendar.getInstance();
- gc.setTime(beginTime);
- XMLGregorianCalendar gTime = null;
- try {
- gTime =
DatatypeFactory.newInstance().newXMLGregorianCalendar((GregorianCalendar)
gc).normalize();
- } catch (DatatypeConfigurationException e) {
- LOG.error("Unable to set the begin time", e);
- }
- gTime.setFractionalSecond(null);
- LOG.info("Begin Time: " + gTime);
- request.setExclusiveBeginTimestamp(gTime);
- }
+ gTime =
DatatypeFactory.newInstance().newXMLGregorianCalendar((GregorianCalendar)
gc).normalize();
+ } catch (DatatypeConfigurationException e) {
+ ErrorUtils.RuntimeErrors.ILLEGAL_STATE.throwRuntime("Unable to
set the begin time due to", e);
+ }
+ gTime.setFractionalSecond(null);
+ LOG.info("Begin Time: " + gTime);
+ request.setExclusiveBeginTimestamp(gTime);
+ }
- try {
- PollResponse response = call(request, PollResponse.class);
- LOG.info("Got Poll Response with " +
response.getContentBlocks().size() + " blocks");
- int numProcessed = 0;
- long avgTimeMS = 0;
- long timeStartedBlock = System.currentTimeMillis();
- for (ContentBlock block : response.getContentBlocks()) {
- AnyMixedContentType content = block.getContent();
- for (Object o : content.getContent()) {
- numProcessed++;
- long timeS = System.currentTimeMillis();
- String xml = null;
- if (o instanceof Element) {
- Element element = (Element) o;
- xml =
getStringFromDocument(element.getOwnerDocument());
- if(LOG.isDebugEnabled() && Math.random() <
0.01) {
- LOG.debug("Random Stix doc: " + xml);
- }
- for (LookupKV<EnrichmentKey, EnrichmentValue>
kv : extractor.extract(xml)) {
- if(allowedIndicatorTypes.isEmpty()
- ||
allowedIndicatorTypes.contains(kv.getKey().type)
- )
- {
-
kv.getValue().getMetadata().put("source_type", "taxii");
-
kv.getValue().getMetadata().put("taxii_url", endpoint.toString());
-
kv.getValue().getMetadata().put("taxii_collection", collection);
- Put p = converter.toPut(columnFamily,
kv.getKey(), kv.getValue());
- HTableInterface table =
getTable(hbaseTable);
- table.put(p);
- LOG.info("Found Threat Intel: " +
kv.getKey() + " => " + kv.getValue());
- }
- }
- }
- avgTimeMS += System.currentTimeMillis() - timeS;
- }
- if( (numProcessed + 1) % 100 == 0) {
- LOG.info("Processed " + numProcessed + " in " +
(System.currentTimeMillis() - timeStartedBlock) + " ms, avg time: " + avgTimeMS
/ content.getContent().size());
- timeStartedBlock = System.currentTimeMillis();
- avgTimeMS = 0;
- numProcessed = 0;
- }
+ try {
+ PollResponse response = call(request, PollResponse.class);
+ LOG.info("Got Poll Response with " +
response.getContentBlocks().size() + " blocks");
+ int numProcessed = 0;
+ long avgTimeMS = 0;
+ long timeStartedBlock = System.currentTimeMillis();
+ for (ContentBlock block : response.getContentBlocks()) {
+ AnyMixedContentType content = block.getContent();
+ for (Object o : content.getContent()) {
+ numProcessed++;
+ long timeS = System.currentTimeMillis();
+ String xml = null;
+ if (o instanceof Element) {
+ Element element = (Element) o;
+ xml = getStringFromDocument(element.getOwnerDocument());
+ if(LOG.isDebugEnabled() && Math.random() < 0.01) {
+ LOG.debug("Random Stix doc: " + xml);
+ }
+ for (LookupKV<EnrichmentKey, EnrichmentValue> kv :
extractor.extract(xml)) {
+ if(allowedIndicatorTypes.isEmpty()
+ || allowedIndicatorTypes.contains(kv.getKey().type)
+ )
+ {
+ kv.getValue().getMetadata().put("source_type", "taxii");
+ kv.getValue().getMetadata().put("taxii_url",
endpoint.toString());
+ kv.getValue().getMetadata().put("taxii_collection",
collection);
+ Put p = converter.toPut(columnFamily, kv.getKey(),
kv.getValue());
+ HTableInterface table = getTable(hbaseTable);
+ table.put(p);
+ LOG.info("Found Threat Intel: " + kv.getKey() + " => " +
kv.getValue());
}
- } catch (Exception e) {
- LOG.error(e.getMessage(), e);
- throw new RuntimeException("Unable to make request", e);
+ }
}
+ avgTimeMS += System.currentTimeMillis() - timeS;
+ }
+ if( (numProcessed + 1) % 100 == 0) {
+ LOG.info("Processed " + numProcessed + " in " +
(System.currentTimeMillis() - timeStartedBlock) + " ms, avg time: " + avgTimeMS
/ content.getContent().size());
+ timeStartedBlock = System.currentTimeMillis();
+ avgTimeMS = 0;
+ numProcessed = 0;
+ }
}
- finally {
- inProgress = false;
- beginTime = ts;
- }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ throw new RuntimeException("Unable to make request", e);
+ }
+ }
+ finally {
+ inProgress = false;
+ beginTime = ts;
}
- public String getStringFromDocument(Document doc)
+ }
+ public String getStringFromDocument(Document doc)
+ {
+ try
{
- try
- {
- DOMSource domSource = new DOMSource(doc);
- StringWriter writer = new StringWriter();
- StreamResult result = new StreamResult(writer);
- TransformerFactory tf = TransformerFactory.newInstance();
- Transformer transformer = tf.newTransformer();
- transformer.transform(domSource, result);
- return writer.toString();
- }
- catch(TransformerException ex)
- {
- ex.printStackTrace();
- return null;
- }
+ DOMSource domSource = new DOMSource(doc);
+ StringWriter writer = new StringWriter();
+ StreamResult result = new StreamResult(writer);
+ TransformerFactory tf = TransformerFactory.newInstance();
+ tf.setFeature(XMLConstants.FEATURE_SECURE_PROCESSING, true);
+ Transformer transformer = tf.newTransformer();
+ transformer.transform(domSource, result);
+ return writer.toString();
}
- private <RESPONSE_T> RESPONSE_T call( Object request,
Class<RESPONSE_T> responseClazz) throws URISyntaxException, JAXBException,
IOException {
- return call(taxiiClient, endpoint.toURI(), request, context,
responseClazz);
+ catch(TransformerException ex)
+ {
+ ex.printStackTrace();
+ return null;
}
+ }
+ private <RESPONSE_T> RESPONSE_T call( Object request, Class<RESPONSE_T>
responseClazz) throws URISyntaxException, JAXBException, IOException {
+ return call(taxiiClient, endpoint.toURI(), request, context,
responseClazz);
+ }
- private void initializeClient(TaxiiConnectionConfig config) throws
Exception {
- LOG.info("Initializing client..");
- if(context == null) {
- context = createContext(config.getEndpoint(),
config.getUsername(), config.getPassword(), config.getPort());
- }
- URL endpoint = config.getEndpoint();
- if(config.getType() == ConnectionType.DISCOVER) {
- LOG.info("Discovering endpoint");
- endpoint = discoverPollingClient(config.getProxy(), endpoint,
config.getUsername(), config.getPassword(), context, collection).pollEndpoint;
- this.endpoint = endpoint;
- LOG.info("Discovered endpoint as " + endpoint);
- }
- taxiiClient = buildClient(config.getProxy(), config.getUsername(),
config.getPassword());
+ private void initializeClient(TaxiiConnectionConfig config) throws
Exception {
+ LOG.info("Initializing client..");
+ if(context == null) {
+ context = createContext(config.getEndpoint(), config.getUsername(),
config.getPassword(), config.getPort());
}
-
- private static class DiscoveryResults {
- URL pollEndpoint;
- URL collectionManagementEndpoint;
- List<String> collections = new ArrayList<>();
+ URL endpoint = config.getEndpoint();
+ if(config.getType() == ConnectionType.DISCOVER) {
+ LOG.info("Discovering endpoint");
+ endpoint = discoverPollingClient(config.getProxy(), endpoint,
config.getUsername(), config.getPassword(), context, collection).pollEndpoint;
+ this.endpoint = endpoint;
+ LOG.info("Discovered endpoint as " + endpoint);
}
- private static DiscoveryResults discoverPollingClient(URL proxy, URL
endpoint, String username, String password, HttpClientContext context, String
defaultCollection) throws Exception {
+ taxiiClient = buildClient(config.getProxy(), config.getUsername(),
config.getPassword());
+ }
- DiscoveryResults results = new DiscoveryResults();
- {
- HttpClient discoverClient = buildClient(proxy, username,
password);
- String sessionID = MessageHelper.generateMessageId();
- // Prepare the message to send.
- DiscoveryRequest request =
messageFactory.get().createDiscoveryRequest()
- .withMessageId(sessionID);
- DiscoveryResponse response = call(discoverClient,
endpoint.toURI(), request, context, DiscoveryResponse.class);
- for (ServiceInstanceType serviceInstance :
response.getServiceInstances()) {
- if (serviceInstance.isAvailable() &&
serviceInstance.getServiceType() == ServiceTypeEnum.POLL) {
- results.pollEndpoint = new
URL(serviceInstance.getAddress());
- }
- else if(serviceInstance.isAvailable() &&
serviceInstance.getServiceType() == ServiceTypeEnum.COLLECTION_MANAGEMENT) {
- results.collectionManagementEndpoint= new
URL(serviceInstance.getAddress());
- }
- }
- if (results.pollEndpoint == null) {
- throw new RuntimeException("Unable to discover a poll
TAXII feed");
- }
+ private static class DiscoveryResults {
+ URL pollEndpoint;
+ URL collectionManagementEndpoint;
+ List<String> collections = new ArrayList<>();
+ }
+ private static DiscoveryResults discoverPollingClient(URL proxy, URL
endpoint, String username, String password, HttpClientContext context, String
defaultCollection) throws Exception {
+
+ DiscoveryResults results = new DiscoveryResults();
+ {
+ HttpClient discoverClient = buildClient(proxy, username, password);
+ String sessionID = MessageHelper.generateMessageId();
+ // Prepare the message to send.
+ DiscoveryRequest request =
messageFactory.get().createDiscoveryRequest()
+ .withMessageId(sessionID);
+ DiscoveryResponse response = call(discoverClient, endpoint.toURI(),
request, context, DiscoveryResponse.class);
+ for (ServiceInstanceType serviceInstance :
response.getServiceInstances()) {
+ if (serviceInstance.isAvailable() &&
serviceInstance.getServiceType() == ServiceTypeEnum.POLL) {
+ results.pollEndpoint = new URL(serviceInstance.getAddress());
}
- if(defaultCollection == null)
- //get collections
- {
- HttpClient discoverClient = buildClient(proxy, username,
password);
- String sessionID = MessageHelper.generateMessageId();
- CollectionInformationRequest request =
messageFactory.get().createCollectionInformationRequest()
-
.withMessageId(sessionID);
- CollectionInformationResponse response = call(discoverClient,
results.collectionManagementEndpoint.toURI(), request, context,
CollectionInformationResponse.class);
- LOG.info("Unable to find the default collection; available
collections are:");
- for(CollectionRecordType c : response.getCollections()) {
- LOG.info(c.getCollectionName());
- results.collections.add(c.getCollectionName());
- }
- System.exit(0);
+ else if(serviceInstance.isAvailable() &&
serviceInstance.getServiceType() == ServiceTypeEnum.COLLECTION_MANAGEMENT) {
+ results.collectionManagementEndpoint= new
URL(serviceInstance.getAddress());
}
- return results;
+ }
+ if (results.pollEndpoint == null) {
+ throw new RuntimeException("Unable to discover a poll TAXII feed");
+ }
}
+ if(defaultCollection == null)
+ //get collections
+ {
+ HttpClient discoverClient = buildClient(proxy, username, password);
+ String sessionID = MessageHelper.generateMessageId();
+ CollectionInformationRequest request =
messageFactory.get().createCollectionInformationRequest()
+ .withMessageId(sessionID);
+ CollectionInformationResponse response = call(discoverClient,
results.collectionManagementEndpoint.toURI(), request, context,
CollectionInformationResponse.class);
+ LOG.info("Unable to find the default collection; available
collections are:");
+ for(CollectionRecordType c : response.getCollections()) {
+ LOG.info(c.getCollectionName());
+ results.collections.add(c.getCollectionName());
+ }
+ System.exit(0);
+ }
+ return results;
+ }
- private static HttpClientContext createContext(URL endpoint, String
username, String password, int port) {
- HttpClientContext context = null;
- HttpHost target = new HttpHost(endpoint.getHost(), port,
endpoint.getProtocol());
- if (username != null && password != null) {
+ private static HttpClientContext createContext(URL endpoint, String
username, String password, int port) {
+ HttpClientContext context = null;
+ HttpHost target = new HttpHost(endpoint.getHost(), port,
endpoint.getProtocol());
+ if (username != null && password != null) {
- CredentialsProvider credsProvider = new
BasicCredentialsProvider();
- credsProvider.setCredentials(
- new AuthScope(target.getHostName(), target.getPort()),
- new UsernamePasswordCredentials(username, password));
+ CredentialsProvider credsProvider = new BasicCredentialsProvider();
+ credsProvider.setCredentials(
+ new AuthScope(target.getHostName(), target.getPort()),
+ new UsernamePasswordCredentials(username, password));
- //
http://hc.apache.org/httpcomponents-client-ga/tutorial/html/authentication.html
- AuthCache authCache = new BasicAuthCache();
- authCache.put(target, new BasicScheme());
+ //
http://hc.apache.org/httpcomponents-client-ga/tutorial/html/authentication.html
+ AuthCache authCache = new BasicAuthCache();
+ authCache.put(target, new BasicScheme());
- // Add AuthCache to the execution context
- context = HttpClientContext.create();
- context.setCredentialsProvider(credsProvider);
- context.setAuthCache(authCache);
- } else {
- context = null;
- }
- return context;
+ // Add AuthCache to the execution context
+ context = HttpClientContext.create();
+ context.setCredentialsProvider(credsProvider);
+ context.setAuthCache(authCache);
+ } else {
+ context = null;
}
+ return context;
+ }
- public static <RESPONSE_T, REQUEST_T> RESPONSE_T call( HttpClient
taxiiClient
- , URI endpoint
- , REQUEST_T request
- , HttpClientContext context
- , Class<RESPONSE_T> responseClazz
- ) throws JAXBException, IOException {
- //TaxiiXml taxiiXml = xmlFactory.get().createTaxiiXml();
- //String req = taxiiXml.marshalToString(request, true);
- // Call the service
- Object responseObj = taxiiClient.callTaxiiService(endpoint,
request, context);
- LOG.info("Request made : " + request.getClass().getCanonicalName()
+ " => " + responseObj.getClass().getCanonicalName() + " (expected " +
responseClazz.getCanonicalName() + ")");
- //String resp = taxiiXml.marshalToString(responseObj, true);
- try {
- return responseClazz.cast(responseObj);
- }
- catch(ClassCastException cce) {
- TaxiiXml taxiiXml = xmlFactory.get().createTaxiiXml();
- String resp = taxiiXml.marshalToString(responseObj, true);
- String msg = "Didn't return the response we expected: " +
responseObj.getClass() + " \n" + resp;
- LOG.error(msg, cce);
- throw new RuntimeException(msg, cce);
- }
+ public static <RESPONSE_T, REQUEST_T> RESPONSE_T call( HttpClient
taxiiClient
+ , URI endpoint
+ , REQUEST_T request
+ , HttpClientContext context
+ , Class<RESPONSE_T> responseClazz
+ ) throws JAXBException, IOException {
+ //TaxiiXml taxiiXml = xmlFactory.get().createTaxiiXml();
--- End diff --
Can we delete this commented code?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
