Hive external tables shows no data

I'm currently working on a project that ingest a lot of security related data (firewall logs, http proxy logs, ...) and the goal is to keep them for a year or so to be able to do analytics and go back in time in case of security incident and have a clue what the "bad guy" did (follow compromised asset or user).

The plan is to store the data in avro format on HDFS and then create Hive table on top, ideally the table should be external so that Hive doesn't remove the original files in case I need it for something else.

To create an external table using avro formatted data you typically do this:

CREATE EXTERNAL TABLE cyber.proxy_logs
PARTITIONED BY (year STRING, month STRING, day STRING)
ROW FORMAT
SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
STORED AS
INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
LOCATION '/cyber/proxy_logs/'
TBLPROPERTIES ('avro.schema.literal'=' { "type" : "record", "name" : "proxy", "fields" : [ { "name" : "DeviceHostname", "type" : "string", "default" : "" }  { "name" : "ResponseTime", "type" : "string", "default" : "" }, { "name" : "SourceIp", "type" : "string", "default" : "" }, { "name" : "SourcePort", "type" : "string", "default" : "" }, { "name" : "Username", "type" : "string", "default" : "" } ] } ');

This will create a table mapped to our data, the files in HDFS are organized like this:

/hadoop/proxy/year=2017/month=09/day=19
/hadoop/proxy/year=2017/month=09/day=20
...

So that we have 3 levels of partitioning making it easy to do operations on yearly, monthly and daily basis or at least that's the plan.
Once you've done that, you need to tell Hive to add your partitions using:

ALTER TABLE proxy_logs ADD PARTITION(year=2017, month="09", day=19);
...

We need to do that every day since our partitioning is done daily, at this point when I ran a

select count(*) from proxy_logs

I didn't get any result at all... turns out you also need to recompute the statistics for the table metadata to be up to date, I did this with

ANALYZE TABLE proxy_logs partition(year=2017, month="09", day=19) compute statistics;

and only then was I able to access my logs, I will now need to do this for every partitions of every table, time to get scripting!

setup 'bro' to log into elasticsearch

A recurring question on the bro mailling list is, "how do you setup bro to ship logs to elasticsearch?"
I'll explain how I've setup my bro network monitor and hopefully it will be useful to others as well.

Setup bro to log in json format:

There are others recipes where you can use grok to parse the text logs to json-ify it, but I find it easier and more stable if the application can log directly into json.

In you local.bro add the following lines:

redef LogAscii::json_timestamps = JSON::TS_ISO8601;
@load tuning/json-logs

The 1st line changes the format of the time stamp from UNIX epoch to Iso 8601, that make it easier for logstash to parse the date into @timestamp, the 2nd line loads a tunning script that will turn your logs into json.

restart bro with broctl restart --clean and you should be set.

Setup logstash input to feed on the logs:

Note that a few assumptions are made here.
I like to organise my logstash conf in different files, named [input, output, filter]-<purpose>.conf, this make it easy for templating and debugging. Off course you can always stuff it into a gigantic file, that entirely up to you :-) and in the output-default.conf, you will need to change the cluster name to whatever name your cluster has. you will also need to change the input path to match yours and so on ...
final gotcha; on Debian you will need to apt-get install logstash-contrib to have the translate plugin.

In /etc/logstash/conf.d/input-bro.conf, we specify the source files, the type and that the logs are in json.

input {
    file {
       type => 'bro_logs' 
       path => [ '/opt/spool/manager/*.log' ]
        codec => "json"
    }
}

In /etc/logstash/conf.d/filter-bro.conf, the translate will add a field named conn_state_full with the full text based on the content of conn_state, the grok will add a field named bro_type with the type of bro logs (conn, weird, dns, ssl ...) based on the file name, we could do that in the input part as well by giving all the file name and adding a specific type but I'm lazy and I always forget one file so ...

filter {
        if [type] == "bro_logs" {
                date { 
                        match => [ "ts", "ISO8601" ]
                }
                translate { field => "conn_state" destination => "conn_state_full" dictionary => [ "S0", "Attempt", "S1", "Established", "S2", "Originator close only", "S3", "Responder close only", "SF", "SYN/FIN completion", "REJ", "Rejected", "RSTO", "Originator aborted", "RSTR", "Responder aborted", "RSTOS0", "Originator SYN +  RST", "RSTRH", "Responder SYN ACK + RST", "SH", "Originator SYN + FIN", "SHR", "Responder SYN ACK + FIN", "OTH", "Midstream traffic" ] }
                grok {
                        match => { "path" => ".*\/(?<bro_type>[a-zA-Z0-9]+)\.log$"} 
                } 
        }
}

In /etc/logstash/conf.d/output-default.conf

output { 
        elasticsearch {
                cluster => 'elasticsearch'
        }
        #stdout { codec => rubydebug } 
}

restart logstash.

Conclusion:

By now, you should have logs in json and logstash shipping them to elasticsearch, you can now start kibana and visualize our bro logs:

pretty pictures ^_^

Honeypot malware collection

The setup

I've been toying with the idea of installing a kippo honeypot in my company network for a while but 6 months ago I had some time to make a basic setup.
By basic, I mean that kippo runs:

• as a captive user
• in a chroot
• listen to a non privileged port (I used iptables forwarding to redirect external traffic targeted to port 22 to kippo's active port) 
• on a host that's blacklisted in the IDS and iptables config of the rest of the machine (to prevent a lateral move)

The non basic version would involve selinux and strong auditing policy, in case you wonder.

All that is to avoid some skilled people to be able to exploit a vulnerability in kippo and move further on in my network.

And then I let it run for a while ...

The deception

It turns out that, kippo is very popular and that means that most scanning scripts have evolved to detect kippo and quickly run away. 

Some scanner will flag you as honeypot if the password is too easy (root/root), some will try to run commands that kippo doesn't support, many use sftp to try to login ...

I had to use this fork instead, note that this fork also take care of json logging (logstash!) and ssh algorithm fingerprint which is also popular

The hit

After switching to a more capable version of kippo, it was like I opened a big malware tap, I collected 2898 samples, most of them (2101) are ELF-32 executable, 382 are 64 bit executable around 200 perl IRC bot and so on, all in all pretty good !

The 10 most common ones account for nearly 2/3 of the total:

183 1a8712007f9ef593044350226b829a9fb25f91ad ==> elf32
186 acbe528883175ce934df4edd4fff045a0e2d2d8f ==> elf32
187 bcf7c4b4621a6452f8ace5e1c0df78c71f7ae4bb ==> "C" source jessica_biel_naked_in_my_bed.c
187 dc063902fc457a2d13b0d91ebc4d508bf6bfd118 ==> elf32
188 ba61480ec4062c3386a7b1e559dab2f0baf5e98f ==> elf32 (always fdsfsfvff, left handed?)
188 ec22fac0510d0dc2c29d56c55ff7135239b0aeee ==> elf32
189 059964612c1ac7c928b81a99da67aed3f3a41865 ==> elf64 (always rewgtf3er4t)
189 44e569a191a5d7bd720c7af06c2fd81a501a245b ==> elf32 to replace udevd
190 0e76f4c72295fe851b775dac8c49ec53108f1df6 ==> elf64
236 27e67a31ffc2797340a02133a4bfab5584faa65d ==> elf32

There is a very popular vmsplice exploit (jessical biel ...), so far nothing spectacular all the sha1's existed on Virus total around the same date I received my 'copy'

The weird

Notable mention, some automated tools assumed that because the username / password combination succeeded the device on which it just log must be the expected ones. 

As an amusing side effect, I now have malware build for mips and ARM:

25dc278de8f8b80cea05e9afc4faac6df9b0638b  ==> MIPS32
2a63299784407db16cd3168a197fe57070b1ff83 ==> ARM
3e73c1a31580a6d0e65c4c8a436ec4be8f00c496 ==> MIPS32
4d90877a832ae21befd5a5556b2bfec3c2404c35  ==> ARM
7130128fc2bcafa4b4fe0ba6159399432d833dfd  ==> MIPS-I
7d0ab04aa3c835956d3fe6549ec7bf0223931468  ==> ARM
7e54ec563e186f225b2d04e0f8f1d28dffdd6fb1  ==> ARM
8103179432bcc189f81810bacd191e4440f3a0a3  ==> ARM
9b3b2c7eecad5ab0bf5cb37f094ceb951d7ae52c  ==> ARM
c421bfebe129aa4179e769b56fccb37bb026a1b2  ==> ARM (not stripped)
db79126d667109c3df6138d55b8669fe1a1f10f4  ==> MIPS32

Targeted to routers, ADSL modems, IP cameras and other embedded devices ...

The end?

I'm still working on that project as time allows, I'm very curious to understand the purpose of the MIPS/ARM malware, I suspect some of them change the dns settings to redirect / mitm the traffic of the appliance but I suspect there's more than that. 

I've been looking for a dtrace alternative on linux for a while and recently came accross sysdig which looks promising.
A few interting one liners:

get all the threads id of a running process (mysqld):

sysdig -p %thread.tid proc.name=mysqld| sort -u

Show the activity of a single thread:

sysdig thread.tid=5445 proc.name=mysqld

The intersting point of sysdig compared to other is the ability to script 'chisel' allowing to write a complex probe, the default install comes with a few

Here's one of mine, getting the top allocating threads of a process:

--[[
Author: Jérémie Banier
Contact: jbanier@gmail.com
Date: 08 Sep 2014
 
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License version 2 as
published by the Free Software Foundation.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see .
--]]

-- The number of items to show
TOP_NUMBER = 30

-- Chisel description
description = "Show the top " .. TOP_NUMBER .. " threads allocating memory. You can use filters to restrict this to a specific process, thread or file."
short_description = "Top threads by memory allocation"
category = "Performance"

-- Chisel argument list
args = {}

-- Argument notification callback
function on_set_arg(name, val)
 return false
end

-- Initialization callback
function on_init()
 chisel.exec("table_generator", 
  "thread.tid",
  "Thread ID",
  "evt.count",
  "# Calls",
  "evt.type=brk",
  "" .. TOP_NUMBER,
  "none")
 return true
end

(save it in ~/.chisels then invoke it with sysdig -c topmemory proc.name=mysqld)

Sadly it still miss a few dtrace feature like enabling a probe when you got into specific user land function or the ability to display a stack trace but the tool is still young :)

There's a JSON output format available which means you could trace all reads and writes, send them to kibana and graph them or trace all syscall failing with ENOMEM, EAGAIN, ... graph and know when you're getting behind in terms of capacity. all kind of cool stuff :)

A journey in the country of Winnie the malware

A couple of weeks ago I've deployed a honeypot in our network perimeter to have an idea of how aggressive the peoples scanning our network really are. I've decided to install kippo a medium interaction honeypot, medium interaction means you can log on and run a set of limited commands and it acts like a *real* machine, it traces everything you do, save everything you downloads ... To make sure the joke wasn't on me, I run kippo in a chroot as a non privileged user it listen on port 2222 and iptables does the forwarding to kippo for those not on my network, pretty neat.

1st catch:

After a few hours, the first connection starts to appears and the 1st users try to login. The funny thing is, they all use sftp and not the expected ssh ? The reason for that is kippo is a popular honeypot and it _doesn't_ support sftp yet, so the attackers use sftp to avoid falling into honeypot ! Luckily a patched version exist that support sftp, once I used that one people stick around a bit more but not that long, the next trick in the attacker sleeve is 'iptables', kippo doesn't implement the command soooo ... you get the idea. Good news again, kippo is easy to extend and simply adding a text file containing the output of the iptables command to "txtcmds/sbin/iptables" is enough to lure some automated scanner into the trap (until next week or so)

Passwords!:

One of the interesting intel to collect is what password do the attacker try ? Well here's a small sample of the most popular passwords you shouldn't use:

     18 [admin/123123]

     18 [admin/1234567890]

     18 [admin/12345678]

     18 [admin/1234]

     18 [admin/123qwe!@#]

     18 [admin/142536]

     18 [admin/1qaz2wsx]

     18 [admin/data]

     18 [admin/qweasd]

     18 [admin/rootme]

     19 [admin/123456]

     19 [admin/P@ssw0rd]

     19 [admin/admin123]

     19 [admin/passw0rd]

     19 [admin/qwe123]

     19 [admin/root123]

     19 [admin/root@123]

     21 [admin/12345]

     22 [admin/password]

     23 [admin/root]

     27 [admin/admin]

     32 [root/root]

    214 [root/admin]

Note the high score of the root/admin combo, the classics never dies or so it seems.

Malware collection:

Another cool feature of kippo, is that it will backup anything the attacker downloads by curl, wget and so on and again pretty quickly you get a few samples so far I have received:

822dd344bfa3ab37ebc968140f5f6296  http___mdb7_cn_8081_star 1.1M

5cdf87129e45d9a3132b7b4840237190 http___121_40_196_12_65533_wawa 834K

I'll try to reverse engineer those samples as time allows (not so much I'm afraid) but I can already provide a few info:

star:

by running 'string' on the 1st sample (star) I find a large list of ip addresses, likely compromised hosts used for C&C:

61.132.163.68

202.102.192.68

202.102.213.68

202.102.200.101

58.242.2.2

202.38.64.1

211.91.88.129

211.138.180.2

218.104.78.2

[...] 

The executable is not stripped and contains lots of mangled symbols indicating that it has been coded in C++, ldd show now external dependencies meaning that for portability it was statically linked, all in all it seems pretty neat !

wawa:

The seconds sample looks a bit more elaborate, like star it is statically linked but this times all symbols have been stripped and running strings on it reveals the following:

$Info: This file is packed with the UPX executable packer http://upx.sf.net $

$Id: UPX 3.91 Copyright (C) 1996-2013 the UPX Team. All Rights Reserved. $

Which means  that it was packed to make the work of potential reverse engineers more difficult but not impossible since UPX is open source. 

I hope to give it a go one of these days and try out http://www.radare.org at the same time, in the mean time if you have details on those malware or want more info on them don't hesitate to drop me a note.

Elasticsearch housekeeping

Logstash is very cool but the underlying Elasticsearch engine can take up a lot of space, so I wrote a small cleaning up script that runs daily to either discard older than 30 days data or optimize active tables.

#!/usr/bin/python

import pycurl
import json
import StringIO
from datetime import datetime, timedelta

retentionDays = 30

c = pycurl.Curl()
b = StringIO.StringIO()

c.setopt(c.URL, 'http://127.0.0.1:9200/_status')
c.setopt(pycurl.WRITEFUNCTION, b.write)
c.perform()

blob = json.loads( b.getvalue() )

for index in blob['indices']:
 if 'logstash' in index:
  old = datetime.now() - timedelta(days = retentionDays)
  indexDate = datetime.strptime(index, "logstash-%Y.%m.%d")
  if old > indexDate:
   print "delete", index
   c.setopt(pycurl.CUSTOMREQUEST, "DELETE")
   c.setopt(c.URL, ('http://127.0.0.1:9200/%s').format(index))
   c.perform()
  else:
   print "optimize", index
   c.setopt(c.URL, ('http://127.0.0.1:9200/%s/_optimize').format(index))
   c.perform()

Turns out there is a much better tool to do all Elasticsearch related housekeeping called curator but anyway sometimes it's nice to make your own scripts :-)

Scapy - Rmcp / ipmi fuzzer

J'avais déjà joué avec scapy  dans un billet précédent et je me suis demandé à quel point c'est difficile de rajouter un nouveau protocole... Suite à ça j'ai cherché un truc raisonnablement simple et sympa et je me suis penché sur Rmcp / ipmi ...
Ipmi d'après Wikipedia est: "L'Interface de gestion intelligente de matériel, (ou IPMI, Intelligent Platform Management Interface) est un ensemble de spécifications d'interfaces communes avec du matériel informatique (principalement des serveurs) permettant de surveiller certains composants (ventilateur, sonde de température, ...), mais également de contrôler l'ordinateur à distance, reboot, interrupteur, console à distance."
En gros ça se présente sous la forme d'un chip intégré à la carte réseau d'un serveur et ça écoute pour recevoir des commande bas niveau de type combien de courant consomme tu ? coupe l'alimentation, effectue un redémarrage par acpi et ainsi de suite ... Les firmwares embarqués sont toujours de bon candidat au fuzzing car pas souvent mis à jour et mis à jour pénible et c'est pas comme si ça n'avait pas posé des problèmes dans le passé ...

#! /usr/bin/env python
#vim: set fileencoding=latin-1
# Author: Jérémie Banier
# Date: Oct. 29 2013
# Purpose: implement / test ipmi protocol with scapy
# Based on test add-ons sample 
# usage:

import logging
# Set log level to benefit from Scapy warnings
logging.getLogger("scapy").setLevel(1)

from scapy.all import *

class Rmcp(Packet):
    name = "Remote Management Control Protocol"
    fields_desc=[ LEShortField("Version",0x06),
        ByteField("Sequence", 0xFF) ,
        XByteField("Type and Class", 0x07) , 
        ByteEnumField("Authentication type", 0, {0:'None', 6:'RMCP+'}), ]

bind_layers( UDP, Rmcp, sport=623 )
bind_layers( UDP, Rmcp, dport=623 )

class IPMISessionLayer(Packet):
    name = "IPMI Session Wrapper"
    fields_desc=[ IntField("Session sequence number", 0),
            XIntField("Session ID", 0),
            ByteField("Message length", 0), ]

class IPMISessionLayer2(Packet):
    name = "IPMI Session Wrapper v2.0+"
    fields_desc=[ ByteEnumField("Payload type", 0x10, 
                {0x10:"Open session request", 0x11:"Open session response",
                    0x12:"RAKP Message 1", 0x13:"RAKP Message 2"
                    }),
            IntField("Session sequence number", 0),
            XIntField("Session ID", 0),
            ByteField("Message length", 0), ]

bind_layers( Rmcp, IPMISessionLayer, {'Authentication type':0} )
bind_layers( Rmcp, IPMISessionLayer2, {'Authentication type':6} )

class IPMILayer(Packet):
    name = "Intelligent Platform Management Interface"
    fields_desc = [ ByteField("Target address", 0x20), ByteEnumField("Target LUN", 0x18, {0x18:"NetFN Application Request"}),
            ByteField("Header checksum", 0xc8), ByteField("Source address", 0x81),
            ByteField("Source LUN", 0x00), 
            ByteEnumField("Command", 0x38, {0x38:"Get Channel Auth. cap."}),
            ByteEnumField("Version compat.", 0x0e, {0x0e:"IPMI v2.0+"}),
            ByteEnumField("Requested privilege level", 0x04, {0x04:"Administrator"}),
            ByteField("Data checksum", 0xb5) ]

bind_layers( IPMISessionLayer, IPMILayer )

if __name__ == "__main__":
    interact(mydict=globals(), mybanner="IPMI fuzzer")

Le script en est encore à l'état d'ébauche mais on peut déjà l'utiliser pour tester l'API de Scapy:

└───> ./Rmcp.py
WARNING: No route found for IPv6 destination :: (no default route?)
Welcome to Scapy (2.2.0)
IPMI fuzzer
>>> t= rdpcap('ipmi2.pcap')
>>> t[16].show()
###[ cooked linux ]###
  pkttype= sent-by-us
  lladdrtype= 0x1
  lladdrlen= 6
  src= '\x00\x1d\tlg,'
  proto= IPv4
###[ IP ]###
     version= 4L
     ihl= 5L
     tos= 0x0
     len= 76
     id= 9063
     flags= DF
     frag= 0L
     ttl= 64
     proto= udp
     chksum= 0x5c0a
     src= 10.200.82.208
     dst= 10.201.82.207
     \options\
###[ UDP ]###
        sport= 41227
        dport= asf_rmcp
        len= 56
        chksum= 0xbb79
###[ Remote Management Control Protocol ]###
           Version= 6
           Sequence= 255
           Type and Class= 0x7
           Authentication type= RMCP+
###[ IPMI Session Wrapper v2.0+ ]###
              Payload type= Open session request
              Session sequence number= 0
              Session ID= 0x0
              Message length= 32

Il me reste encore à implémenter l'ouverture de session, ce qui permettrais de faire du brute force sur le mot de passe admin par exemple (encore qu'il faudrait pour cela que le chip ipmi soit accessible depuis internet ce qui serait pas un très bonne idée pour dire ça poliment)
Une affaire à suivre avec heureusement des détails croustillants :P et une fin heureuse ...