Permalink
Browse files

rec: Log outgoing queries / incoming responses via protobuf

(cherry picked from commit 4898a34)
  • Loading branch information...
rgacogne committed Nov 4, 2016
1 parent 9388f1b commit 7a2a645db683f1ea75a2416b6a1dcc4233e1731c
View
@@ -30,11 +30,10 @@ def run(self):
self.printQueryMessage(msg)
elif msg.type == dnsmessage_pb2.PBDNSMessage.DNSResponseType:
self.printResponseMessage(msg)
- # PR #3869
- # elif msg.type == dnsmessage_pb2.PBDNSMessage.DNSOutgoingQueryType:
- # self.printOutgoingQueryMessage(msg)
- # elif msg.type == dnsmessage_pb2.PBDNSMessage.DNSIncomingResponseType:
- # self.printIncomingResponseMessage(msg)
+ elif msg.type == dnsmessage_pb2.PBDNSMessage.DNSOutgoingQueryType:
+ self.printOutgoingQueryMessage(msg)
+ elif msg.type == dnsmessage_pb2.PBDNSMessage.DNSIncomingResponseType:
+ self.printIncomingResponseMessage(msg)
else:
print('Discarding unsupported message type %d' % (msg.type))
@@ -137,9 +136,9 @@ def printSummary(self, msg, typestr):
messageidstr = binascii.hexlify(bytearray(msg.messageId))
initialrequestidstr = ''
- # PR #3869
- # if msg.HasField('initialRequestId'):
- # initialrequestidstr = ', initial uuid: ' + binascii.hexlify(bytearray(msg.initialRequestId))
+ if msg.HasField('initialRequestId'):
+ initialrequestidstr = ', initial uuid: ' + binascii.hexlify(bytearray(msg.initialRequestId))
+
requestorstr = ''
requestor = self.getRequestorSubnet(msg)
if requestor:
@@ -544,6 +544,22 @@ The optional parameters are:
* asyncConnect = if set to false (default) the first connection to the server during startup will block up to `timeout` seconds,
otherwise the connection is done in a separate thread.
+While `protobufServer()` only exports the queries sent to the recursor from clients, with the corresponding responses,
+`outgoingProtobufServer()` can be used to export outgoing queries sent by the recursor to authoritative servers,
+along with the corresponding responses.
+
+```
+outgoingProtobufServer("192.0.2.1:4242" [[[[, timeout], maxQueuedEntries], reconnectWaitTime], asynConnect])
+```
+
+The optional parameters for `outgoingProtobufServer()` are:
+
+* timeout = time in seconds to wait when sending a message, default to 2
+* maxQueuedEntries = how many entries will be kept in memory if the server becomes unreachable, default to 100
+* reconnectWaitTime = how long to wait, in seconds, between two reconnection attempts, default to 1
+* asyncConnect = if set to false (default) the first connection to the server during startup will block up to `timeout` seconds,
+otherwise the connection is done in a separate thread.
+
The protocol buffers message types can be found in the [`dnsmessage.proto`](https://github.com/PowerDNS/pdns/blob/master/pdns/dnsmessage.proto) file.
## `lua-dns-script`
View
@@ -22,9 +22,7 @@
#include "config.h"
#include "dnsdist.hh"
-#include "gettime.hh"
-#include "dnsparser.hh"
#include "dnsdist-protobuf.hh"
#ifdef HAVE_PROTOBUF
@@ -33,10 +31,7 @@
DNSDistProtoBufMessage::DNSDistProtoBufMessage(DNSProtoBufMessageType type, const DNSQuestion& dq): DNSProtoBufMessage(type, dq.uniqueId, dq.remote, dq.local, *dq.qname, dq.qtype, dq.qclass, dq.dh->id, dq.tcp, dq.len)
{
if (type == Response) {
- PBDNSMessage_DNSResponse* response = d_message.mutable_response();
- if (response) {
- response->set_rcode(dq.dh->rcode);
- }
+ setResponseCode(dq.dh->rcode);
addRRsFromPacket((const char*) dq.dh, dq.len);
}
};
View
@@ -25,6 +25,8 @@ message PBDNSMessage {
enum Type {
DNSQueryType = 1;
DNSResponseType = 2;
+ DNSOutgoingQueryType = 3;
+ DNSIncomingResponseType = 4;
}
enum SocketFamily {
INET = 1; // IPv4 (RFC 791)
@@ -72,4 +74,5 @@ message PBDNSMessage {
optional DNSResponse response = 13;
optional bytes originalRequestorSubnet = 14; // EDNS Client Subnet value
optional string requestorId = 15; // Username of the requestor
+ optional bytes initialRequestId = 16; // UUID of the incoming query that initiated this outgoing query or incoming response
}
View
@@ -27,6 +27,7 @@
#include "iputils.hh"
#include "misc.hh"
+#include "protobuf.hh"
#include "dns.hh"
#include "dnspcap.hh"
#include "dnsparser.hh"
View
@@ -48,21 +48,60 @@
#include "validate-recursor.hh"
#include "ednssubnet.hh"
+#ifdef HAVE_PROTOBUF
+
+static void logOutgoingQuery(std::shared_ptr<RemoteLogger> outgoingLogger, boost::optional<const boost::uuids::uuid&> initialRequestId, const boost::uuids::uuid& uuid, const ComboAddress& ip, const DNSName& domain, int type, uint16_t qid, bool doTCP, size_t bytes)
+{
+ if(!outgoingLogger)
+ return;
+
+ RecProtoBufMessage message(DNSProtoBufMessage::OutgoingQuery, uuid, nullptr, &ip, domain, type, QClass::IN, qid, doTCP, bytes);
+ if (initialRequestId) {
+ message.setInitialRequestID(*initialRequestId);
+ }
+
+// cerr <<message.toDebugString()<<endl;
+ std::string str;
+ message.serialize(str);
+ outgoingLogger->queueData(str);
+}
+
+static void logIncomingResponse(std::shared_ptr<RemoteLogger> outgoingLogger, boost::optional<const boost::uuids::uuid&> initialRequestId, const boost::uuids::uuid& uuid, const ComboAddress& ip, const DNSName& domain, int type, uint16_t qid, bool doTCP, size_t bytes, int rcode, const std::vector<DNSRecord>& records, const struct timeval& queryTime)
+{
+ if(!outgoingLogger)
+ return;
+
+ RecProtoBufMessage message(DNSProtoBufMessage::IncomingResponse, uuid, nullptr, &ip, domain, type, QClass::IN, qid, doTCP, bytes);
+ if (initialRequestId) {
+ message.setInitialRequestID(*initialRequestId);
+ }
+ message.setQueryTime(queryTime.tv_sec, queryTime.tv_usec);
+ message.setResponseCode(rcode);
+ message.addRRs(records);
+
+// cerr <<message.toDebugString()<<endl;
+ std::string str;
+ message.serialize(str);
+ outgoingLogger->queueData(str);
+}
+#endif /* HAVE_PROTOBUF */
+
//! returns -2 for OS limits error, -1 for permanent error that has to do with remote **transport**, 0 for timeout, 1 for success
/** lwr is only filled out in case 1 was returned, and even when returning 1 for 'success', lwr might contain DNS errors
Never throws!
*/
-int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool doTCP, bool sendRDQuery, int EDNS0Level, struct timeval* now, boost::optional<Netmask>& srcmask, LWResult *lwr)
+int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool doTCP, bool sendRDQuery, int EDNS0Level, struct timeval* now, boost::optional<Netmask>& srcmask, boost::optional<const ResolveContext&> context, std::shared_ptr<RemoteLogger> outgoingLogger, LWResult *lwr)
{
size_t len;
size_t bufsize=g_outgoingEDNSBufsize;
scoped_array<unsigned char> buf(new unsigned char[bufsize]);
vector<uint8_t> vpacket;
// string mapped0x20=dns0x20(domain);
+ uint16_t qid = dns_random(0xffff);
DNSPacketWriter pw(vpacket, domain, type);
pw.getHeader()->rd=sendRDQuery;
- pw.getHeader()->id=dns_random(0xffff);
+ pw.getHeader()->id=qid;
/* RFC 6840 section 5.9:
* This document further specifies that validating resolvers SHOULD set
* the CD bit on every upstream query. This is regardless of whether
@@ -98,20 +137,31 @@ int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool d
DTime dt;
dt.set();
*now=dt.getTimeval();
+
+#ifdef HAVE_PROTOBUF
+ boost::uuids::uuid uuid;
+ const struct timeval queryTime = *now;
+
+ if (outgoingLogger) {
+ uuid = (*t_uuidGenerator)();
+ logOutgoingQuery(outgoingLogger, context ? context->d_initialRequestId : boost::none, uuid, ip, domain, type, qid, doTCP, vpacket.size());
+ }
+#endif
+
errno=0;
if(!doTCP) {
int queryfd;
if(ip.sin4.sin_family==AF_INET6)
g_stats.ipv6queries++;
- if((ret=asendto((const char*)&*vpacket.begin(), vpacket.size(), 0, ip, pw.getHeader()->id,
+ if((ret=asendto((const char*)&*vpacket.begin(), vpacket.size(), 0, ip, qid,
domain, type, &queryfd)) < 0) {
return ret; // passes back the -2 EMFILE
}
// sleep until we see an answer to this, interface to mtasker
- ret=arecvfrom(reinterpret_cast<char *>(buf.get()), bufsize-1,0, ip, &len, pw.getHeader()->id,
+ ret=arecvfrom(reinterpret_cast<char *>(buf.get()), bufsize-1,0, ip, &len, qid,
domain, type, queryfd, now);
}
else {
@@ -178,6 +228,11 @@ int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool d
lwr->d_rcode=mdp.d_header.rcode;
if(mdp.d_header.rcode == RCode::FormErr && mdp.d_qname.empty() && mdp.d_qtype == 0 && mdp.d_qclass == 0) {
+#ifdef HAVE_PROTOBUF
+ if(outgoingLogger) {
+ logIncomingResponse(outgoingLogger, context ? context->d_initialRequestId : boost::none, uuid, ip, domain, type, qid, doTCP, len, lwr->d_rcode, lwr->d_records, queryTime);
+ }
+#endif
return 1; // this is "success", the error is set in lwr->d_rcode
}
@@ -210,13 +265,23 @@ int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool d
}
}
+#ifdef HAVE_PROTOBUF
+ if(outgoingLogger) {
+ logIncomingResponse(outgoingLogger, context ? context->d_initialRequestId : boost::none, uuid, ip, domain, type, qid, doTCP, len, lwr->d_rcode, lwr->d_records, queryTime);
+ }
+#endif
return 1;
}
catch(std::exception &mde) {
if(::arg().mustDo("log-common-errors"))
L<<Logger::Notice<<"Unable to parse packet from remote server "<<ip.toString()<<": "<<mde.what()<<endl;
lwr->d_rcode = RCode::FormErr;
- g_stats.serverParseError++;
+ g_stats.serverParseError++;
+#ifdef HAVE_PROTOBUF
+ if(outgoingLogger) {
+ logIncomingResponse(outgoingLogger, context ? context->d_initialRequestId : boost::none, uuid, ip, domain, type, qid, doTCP, len, lwr->d_rcode, lwr->d_records, queryTime);
+ }
+#endif
return 1; // success - oddly enough
}
catch(...) {
View
@@ -40,6 +40,8 @@
#include "pdnsexception.hh"
#include "dns.hh"
#include "namespaces.hh"
+#include "remote_logger.hh"
+#include "resolve-context.hh"
int asendto(const char *data, size_t len, int flags, const ComboAddress& ip, uint16_t id,
const DNSName& domain, uint16_t qtype, int* fd);
@@ -65,5 +67,5 @@ public:
bool d_haveEDNS{false};
};
-int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool doTCP, bool sendRDQuery, int EDNS0Level, struct timeval* now, boost::optional<Netmask>& srcmask, LWResult* res);
+int asyncresolve(const ComboAddress& ip, const DNSName& domain, int type, bool doTCP, bool sendRDQuery, int EDNS0Level, struct timeval* now, boost::optional<Netmask>& srcmask, boost::optional<const ResolveContext&> context, std::shared_ptr<RemoteLogger> outgoingLogger, LWResult* res);
#endif // PDNS_LWRES_HH
View
@@ -735,6 +735,9 @@ void startDoResolve(void *p)
// Ignore the client-set CD flag
pw.getHeader()->cd=0;
}
+#ifdef HAVE_PROTOBUF
+ sr.d_initialRequestId = dc->d_uuid;
+#endif
bool tracedQuery=false; // we could consider letting Lua know about this too
bool variableAnswer = false;
@@ -983,7 +986,11 @@ void startDoResolve(void *p)
L<<Logger::Warning<<"Starting validation of answer to "<<dc->d_mdp.d_qname<<"|"<<QType(dc->d_mdp.d_qtype).getName()<<" for "<<dc->d_remote.toStringWithPort()<<endl;
}
- auto state=validateRecords(ret);
+ ResolveContext ctx;
+#ifdef HAVE_PROTOBUF
+ ctx.d_initialRequestId = dc->d_uuid;
+#endif
+ auto state=validateRecords(ctx, ret);
if(state == Secure) {
if(sr.doLog()) {
L<<Logger::Warning<<"Answer to "<<dc->d_mdp.d_qname<<"|"<<QType(dc->d_mdp.d_qtype).getName()<<" for "<<dc->d_remote.toStringWithPort()<<" validates correctly"<<endl;
@@ -1353,9 +1360,11 @@ void handleRunningTCPQuestion(int fd, FDMultiplexer::funcparam_t& var)
}
}
#ifdef HAVE_PROTOBUF
- if(luaconfsLocal->protobufServer) {
+ if(luaconfsLocal->protobufServer || luaconfsLocal->outgoingProtobufServer) {
dc->d_uuid = (*t_uuidGenerator)();
+ }
+ if(luaconfsLocal->protobufServer) {
try {
const struct dnsheader* dh = (const struct dnsheader*) conn->data;
dc->d_ednssubnet = ednssubnet;
@@ -1455,7 +1464,9 @@ string* doProcessUDPQuestion(const std::string& question, const ComboAddress& fr
boost::uuids::uuid uniqueId;
auto luaconfsLocal = g_luaconfs.getLocal();
if (luaconfsLocal->protobufServer) {
+ uniqueId = (*t_uuidGenerator)();
needECS = true;
+ } else if (luaconfsLocal->outgoingProtobufServer) {
uniqueId = (*t_uuidGenerator)();
}
#endif
@@ -1576,7 +1587,7 @@ string* doProcessUDPQuestion(const std::string& question, const ComboAddress& fr
dc->d_tcp=false;
dc->d_policyTags = policyTags;
#ifdef HAVE_PROTOBUF
- if (luaconfsLocal->protobufServer) {
+ if (luaconfsLocal->protobufServer || luaconfsLocal->outgoingProtobufServer) {
dc->d_uuid = uniqueId;
}
dc->d_ednssubnet = ednssubnet;
@@ -3140,7 +3151,8 @@ int getRootNS(void) {
try {
res=sr.beginResolve(DNSName("."), QType(QType::NS), 1, ret);
if (g_dnssecmode != DNSSECMode::Off && g_dnssecmode != DNSSECMode::ProcessNoValidate) {
- auto state = validateRecords(ret);
+ ResolveContext ctx;
+ auto state = validateRecords(ctx, ret);
if (state == Bogus)
throw PDNSException("Got Bogus validation result for .|NS");
}
View
@@ -1,15 +1,37 @@
+#include "gettime.hh"
+#include "dnsparser.hh"
#include "protobuf.hh"
#include "dnsparser.hh"
#include "gettime.hh"
-DNSProtoBufMessage::DNSProtoBufMessage(DNSProtoBufMessageType type)
+void DNSProtoBufMessage::setType(DNSProtoBufMessageType type)
{
#ifdef HAVE_PROTOBUF
- d_message.set_type(type == DNSProtoBufMessage::DNSProtoBufMessageType::Query ? PBDNSMessage_Type_DNSQueryType : PBDNSMessage_Type_DNSResponseType);
+ switch(type) {
+ case DNSProtoBufMessage::DNSProtoBufMessageType::Query:
+ d_message.set_type(PBDNSMessage_Type_DNSQueryType);
+ break;
+ case DNSProtoBufMessage::DNSProtoBufMessageType::Response:
+ d_message.set_type(PBDNSMessage_Type_DNSResponseType);
+ break;
+ case DNSProtoBufMessage::DNSProtoBufMessageType::OutgoingQuery:
+ d_message.set_type(PBDNSMessage_Type_DNSOutgoingQueryType);
+ break;
+ case DNSProtoBufMessage::DNSProtoBufMessageType::IncomingResponse:
+ d_message.set_type(PBDNSMessage_Type_DNSIncomingResponseType);
+ break;
+ default:
+ throw std::runtime_error("Unsupported protobuf type: "+std::to_string(type));
+ }
#endif /* HAVE_PROTOBUF */
}
+DNSProtoBufMessage::DNSProtoBufMessage(DNSProtoBufMessageType type)
+{
+ setType(type);
+}
+
void DNSProtoBufMessage::setQuestion(const DNSName& qname, uint16_t qtype, uint16_t qclass)
{
#ifdef HAVE_PROTOBUF
@@ -204,6 +226,13 @@ void DNSProtoBufMessage::setUUID(const boost::uuids::uuid& uuid)
std::copy(uuid.begin(), uuid.end(), messageId->begin());
}
+void DNSProtoBufMessage::setInitialRequestID(const boost::uuids::uuid& uuid)
+{
+ std::string* messageId = d_message.mutable_initialrequestid();
+ messageId->resize(uuid.size());
+ std::copy(uuid.begin(), uuid.end(), messageId->begin());
+}
+
void DNSProtoBufMessage::update(const boost::uuids::uuid& uuid, const ComboAddress* requestor, const ComboAddress* responder, bool isTCP, uint16_t id)
{
struct timespec ts;
@@ -213,7 +242,13 @@ void DNSProtoBufMessage::update(const boost::uuids::uuid& uuid, const ComboAddre
setUUID(uuid);
d_message.set_id(ntohs(id));
- d_message.set_socketfamily((requestor && requestor->sin4.sin_family == AF_INET) ? PBDNSMessage_SocketFamily_INET : PBDNSMessage_SocketFamily_INET6);
+ if (requestor) {
+ d_message.set_socketfamily(requestor->sin4.sin_family == AF_INET ? PBDNSMessage_SocketFamily_INET : PBDNSMessage_SocketFamily_INET6);
+ }
+ else if (responder) {
+ d_message.set_socketfamily(responder->sin4.sin_family == AF_INET ? PBDNSMessage_SocketFamily_INET : PBDNSMessage_SocketFamily_INET6);
+ }
+
d_message.set_socketprotocol(isTCP ? PBDNSMessage_SocketProtocol_TCP : PBDNSMessage_SocketProtocol_UDP);
if (responder) {
@@ -225,11 +260,11 @@ void DNSProtoBufMessage::update(const boost::uuids::uuid& uuid, const ComboAddre
}
-DNSProtoBufMessage::DNSProtoBufMessage(DNSProtoBufMessageType type, const boost::uuids::uuid& uuid, const ComboAddress* requestor, const ComboAddress* to, const DNSName& domain, int qtype, uint16_t qclass, uint16_t qid, bool isTCP, size_t bytes)
+DNSProtoBufMessage::DNSProtoBufMessage(DNSProtoBufMessageType type, const boost::uuids::uuid& uuid, const ComboAddress* requestor, const ComboAddress* responder, const DNSName& domain, int qtype, uint16_t qclass, uint16_t qid, bool isTCP, size_t bytes)
{
- update(uuid, requestor, to, isTCP, qid);
+ update(uuid, requestor, responder, isTCP, qid);
- d_message.set_type(type == DNSProtoBufMessage::DNSProtoBufMessageType::Query ? PBDNSMessage_Type_DNSQueryType : PBDNSMessage_Type_DNSResponseType);
+ setType(type);
setBytes(bytes);
setQuestion(domain, qtype, qclass);
Oops, something went wrong.

0 comments on commit 7a2a645

Please sign in to comment.