[
https://issues.apache.org/jira/browse/NIFI-2341?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15445965#comment-15445965
]
ASF GitHub Bot commented on NIFI-2341:
--------------------------------------
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/785#discussion_r76612149
--- Diff:
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ParseCEF.java
---
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.nifi.processors.standard;
+
+import com.fluenda.parcefone.event.CEFHandlingException;
+import com.fluenda.parcefone.event.CommonEvent;
+import com.fluenda.parcefone.parser.CEFParser;
+
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SideEffectFree;
+import org.apache.nifi.annotation.behavior.SupportsBatching;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.stream.io.StreamUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+@EventDriven
+@SideEffectFree
+@SupportsBatching
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Tags({"logs", "cef", "attributes", "system", "event", "message"})
+@CapabilityDescription("Parses the contents of a CEF formatted message and
adds attributes to the FlowFile for " +
+ "headers and extensions of the parts of the CEF message.\n" +
+ "Note: This Processor expects CEF messages WITHOUT the syslog
headers (i.e. starting at \"CEF:0\"")
+@WritesAttributes({@WritesAttribute(attribute = "cef.header.version",
description = "The version of the CEF message."),
+ @WritesAttribute(attribute = "cef.header.deviceVendor", description =
"The Device Vendor of the CEF message."),
+ @WritesAttribute(attribute = "cef.header.deviceProduct", description =
"The deviceProduct of the CEF message."),
+ @WritesAttribute(attribute = "cef.header.deviceVersion", description =
"The deviceVersion of the CEF message."),
+ @WritesAttribute(attribute = "cef.header.deviceEventClassId",
description = "The deviceEventClassId of the CEF message."),
+ @WritesAttribute(attribute = "cef.header.name", description = "The
name of the CEF message."),
+ @WritesAttribute(attribute = "cef.header.severity", description = "The
severity of the CEF message."),
+ @WritesAttribute(attribute = "cef.extension.*", description = "The key
and value generated by the parsing of the message.")})
+@SeeAlso({ParseSyslog.class})
+
+public class ParseCEF extends AbstractProcessor {
+
+
+
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("Any FlowFile that could not be parsed as a CEF
message will be transferred to this Relationship without any attributes being
added")
+ .build();
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Any FlowFile that is successfully parsed as a CEF
message will be to this Relationship.")
+ .build();
+
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_FAILURE);
+ relationships.add(REL_SUCCESS);
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final
ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final CEFParser parser = new CEFParser();
+ final byte[] buffer = new byte[(int) flowFile.getSize()];
+ session.read(flowFile, new InputStreamCallback() {
+ @Override
+ public void process(final InputStream in) throws IOException {
+ StreamUtils.fillBuffer(in, buffer);
+ }
+ });
+
+ CommonEvent event = null;
+
+ try {
+ event = parser.parse(buffer, true);
+ } catch (Exception e) {
+ // This should never trigger but adding in here as a fencing
mechanism to
+ // address possible ParCEFone bugs.
+ getLogger().error("Parser returned unexpected Exception {}
while processing {}; routing to failure", new Object[] {e, flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ }
+
+
+ // ParCEFone returns null every time it cannot parse an
+ // event, so we test
+ if (event==null) {
+ getLogger().error("Failed to parse {} as a CEF message: it
does not conform to the CEF standard; routing to failure", new Object[]
{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ final Map<String, Object> headerFieldMap;
+ final Map<String, Object> extensionFieldMap;
+ try {
+ headerFieldMap = event.getHeader();
+ extensionFieldMap = event.getExtension(true);
+ } catch (CEFHandlingException e) {
+ getLogger().error("Failed to parse {} as a CEF message due to
{}; routing to failure", new Object[] {flowFile, e});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+
+ final Map<String, String> attributes = new HashMap<>();
+
+ // Process KVs of the Header field
+ for (Map.Entry<String, Object> entry : headerFieldMap.entrySet()) {
+ attributes.put("cef.header."+entry.getKey(),
prettyResult(entry.getValue()));
+ }
+
+ for (Map.Entry<String, Object> entry :
extensionFieldMap.entrySet()) {
+ attributes.put("cef.extension." + entry.getKey(),
prettyResult(entry.getValue()));
+ }
+
+ flowFile = session.putAllAttributes(flowFile, attributes);
+ session.transfer(flowFile, REL_SUCCESS);
+ }
+
+ private String prettyResult(Object entryValue) {
+
+ if (entryValue instanceof InetAddress ) {
+ return ((InetAddress) entryValue).getHostAddress();
+ } else if (entryValue instanceof Date) {
+ // Not to discuss. THere should be no date format other than
internationally agreed formats...
+ final SimpleDateFormat formatter = new
SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
+ // Discuss this
+ // formatter.setTimeZone(TimeZone.getTimeZone("UTC"));
--- End diff --
Perhaps this could be a processor property for Local or Absolute time?
> Create a processor to parse logs formated using CEF
> ---------------------------------------------------
>
> Key: NIFI-2341
> URL: https://issues.apache.org/jira/browse/NIFI-2341
> Project: Apache NiFi
> Issue Type: Improvement
> Reporter: Andre
> Assignee: Andre
>
> As NiFi continue to increase its abilities to complement SIEM, Splunk and ELK
> deployments, a number of users will be looking to parse CEF formatted
> logs[1][2].
> CEF is a format specified by Arcsight (now part of HPE) and is described in
> detail in here:
> https://www.protect724.hpe.com/docs/DOC-1072
> [1]
> http://apache-nifi.1125220.n5.nabble.com/Suggestion-of-processors-td9795.html
> [2]
> https://community.hortonworks.com/questions/43185/which-processor-is-used-to-parse-cef-format-logs.html
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
