ip_tools.erl

-module(ip_tools).

-export([next/1, first/2, last/2, from_str/1, to_str/1, range/2]).

next({A,B,C,D}) when D < 255 -> {A,B,C,D+1};
next({A,B,C,_D}) when C < 255 -> {A,B,C+1,0};
next({A,B,_C,_D}) when B < 255 -> {A,B+1,0,0};
next({A,_B,_C,_D}) when A < 255 -> {A+1,0,0,0}.

first(Tuple, Bits) when is_tuple(Tuple) ->
        list_to_tuple(first(tuple_to_list(Tuple), Bits));
first([Next | Rest], Bits) when Bits > 8 -> 
        [Next | first(Rest, Bits - 8)]; 
first([Next | Rest], Bits) -> 
        [Next band (bnot ((1 bsl (8 - Bits)) - 1)) | [0 || _ <- Rest]].

last(Tuple, Bits) when is_tuple(Tuple) ->
        list_to_tuple(last(tuple_to_list(Tuple), Bits));
last([Next | Rest], Bits) when Bits > 8 ->
        [Next | last(Rest, Bits - 8)];
last([Next | Rest], Bits) ->
        [Next bor ((1 bsl (8 - Bits)) - 1) | [255 || _ <- Rest]].

from_str(Str) ->
    [A,B,C,D] = [list_to_integer(X) || X <- string:tokens(Str, ".")],
    {A,B,C,D}.
to_str(Ip) ->
    string:join([integer_to_list(X) || X <- tuple_to_list(Ip)], ".").

range(From, To) when From =:= To -> [From];
range(From, To) ->
    [From | range(next(From), To)].

scanscan.erl

#!/usr/bin/env escript
%%! -smp enable verbose +K true +A 16

-mode(compile).

main([Cidr | Rest]) ->
    [application:start(X) || X <- [inets,crypto,asn1,public_key,ssh]],
    {ok, _} = ssh_http:start_link("yolog.zones.eait.uq.edu.au", 22, "user", "..."),
    Pid = spawn_link(fun bulker_loop/0),
    true = register(bulker, Pid),
    do_range([Cidr | Rest]);

main(_) ->
    io:format("usage: scanscan.erl <cidr> [cidr cidr ...]\n"),
    io:format("scans for things and uploads to elasticsearch\n"),
    halt(1).

do_range([]) -> ok;
do_range([Cidr | Rest]) ->
    [IpStr, BitsStr] = string:tokens(Cidr, "/"),
    Ip = ip_tools:from_str(IpStr),
    Bits = list_to_integer(BitsStr),
    First = ip_tools:first(Ip, Bits),
    Last = ip_tools:last(Ip, Bits),

    io:format("[scanning from ~w to ~w]\n", [First, Last]),

    GS = {fun (I) when I =:= Last -> done; (I) -> ip_tools:next(I) end, First},
    pmap(fun(I = {_, _, _, N}) -> 
        if ((N rem 32) == 0) ->
            io:format("\rdone up to ~-30w", [I]);
        true -> ok end,
        check(I)
    end, GS, 512),
    io:format("\n"),

    do_range(Rest).

check(Ip) ->
    Data = thread([
        fun(H) ->
            TS = os:timestamp(),
            lists:keymerge(1, H, [
                {ip, list_to_binary(ip_tools:to_str(Ip))},
                {timestamp, list_to_binary(ts_to_jstime(TS))}
            ])
        end,
        fun(H) ->
            T1 = os:timestamp(),
            Opts = [{nameservers, [{Ip, 53}]}, {recurse, true}, {timeout, 1000}, {retry, 2}],
            lists:keymerge(1, H, case inet_res:lookup("www.google.com", in, a, Opts) of
                [] -> [{recursive_dns, [{open, false}]}];
                _Res ->
                    T2 = os:timestamp(),
                    [{recursive_dns, [{open, true}, {probe_time, timer:now_diff(T2, T1)}]}]
            end)
        end,
        fun(H) ->
            lists:keymerge(1, H, [{ssh, port_banner(Ip, 22)}])
        end,
        fun(H) ->
            lists:keymerge(1, H, [{vnc, [
                {'0', vnc_banner(Ip, 5900)},
                {'1', vnc_banner(Ip, 5901)}]}])
        end,
        fun(H) ->
            lists:keymerge(1, H, [{'tcp32764', port_banner(Ip, 32764, 500)}])
        end,
        fun(H) ->
            lists:keymerge(1, H, case gen_tcp:connect(Ip, 23, [binary, {active, true}], 1000) of
                {ok, Sock} -> [{telnet, recv_banner(Sock, <<>>, false)}];
                _ -> [{telnet, [{open, false}]}]
            end)
        end,
        fun(H) ->
            lists:keymerge(1, H, case gen_tcp:connect(Ip, 3306, [binary, {active, true}], 1000) of
                {ok, Sock} -> [{mysql, recv_mysql(Sock, <<>>)}];
                _ -> [{mysql, [{open, false}]}]
            end)
        end,
        fun(H) ->
            case inet_res:lookup(Ip, in, ptr, [{timeout, 500}, {retry, 2}]) of
                [Host | _] when is_list(Host) -> [{hostname, list_to_binary(Host)} | H];
                _ -> H
            end
        end
    ], []),
    Json = to_json(Data),
    bulker ! {add_data, self(), Ip, Json},
    receive {bulker, ok} -> ok end.

port_banner(Ip, Port) -> port_banner(Ip, Port, 1000).
port_banner(Ip, Port, Timeout) ->
    case gen_tcp:connect(Ip, Port, [binary, {packet, line}, {active, once}], Timeout) of
        {ok, Sock} ->
            receive
                {tcp, Sock, Line} ->
                    gen_tcp:close(Sock),
                    [{open, true}, {banner, Line}]
            after Timeout ->
                gen_tcp:close(Sock),
                [{open, true}]
            end;
        _ -> [{open, false}]
    end.

vnc_banner(Ip, Port) ->
    case gen_tcp:connect(Ip, Port, [binary, {packet, line}, {active, once}], 1000) of
        {ok, Sock} ->
            receive
                {tcp, Sock, Banner} ->
                    ok = gen_tcp:send(Sock, <<"RFB 003.003\n">>),
                    ok = inet:setopts(Sock, [{packet, raw}]),
                    R = case gen_tcp:recv(Sock, 4) of
                        {ok, <<0:32/big>>} ->
                            {ok, <<ReasonLen:32/big>>} = gen_tcp:recv(Sock, 4),
                            {ok, ReasonBin} = gen_tcp:recv(Sock, ReasonLen),
                            [{reject_reason, ReasonBin}];
                        {ok, <<1:32/big>>} -> [{open, true}, {banner, Banner}, {required_auth, false}];
                        {ok, <<N:32/big>>} when N > 1 -> [{open, true}, {banner, Banner}, {required_auth, true}];
                        _ -> [{open, false}]
                    end,
                    gen_tcp:close(Sock), R
            after 1000 ->
                gen_tcp:close(Sock),
                [{open, false}]
            end;
        _ -> [{open, false}]
    end.

recv_mysql(Sock, Buffer) ->
    receive
        {tcp, Sock, Data} ->
            NewBuffer = <<Buffer/binary, Data/binary>>,
            case NewBuffer of
                <<Len:32/little, Pkt:Len/binary-unit:8, _/binary>> ->
                    gen_tcp:close(Sock),
                    case Pkt of
                        <<10, Info/binary>> ->
                            [Sign, Rest] = binary:split(Info, <<0>>),
                            <<Id:32/little, _Rest2/binary>> = Rest,
                            [{open, true}, {version, Sign}, {id, Id}];
                        _ -> [{open, false}]
                    end;
                _ when byte_size(NewBuffer) < 4096 ->
                    recv_mysql(Sock, NewBuffer);
                _ ->
                    gen_tcp:close(Sock),
                    [{open, false}]
            end;
        {tcp_closed, Sock} ->
            [{open, false}]
    after 2000 ->
        gen_tcp:close(Sock),
        [{open, false}]
    end.

-define(IAC, 255).
-define(TN_WILL, 251).
-define(TN_WONT, 252).
-define(TN_DO, 253).
-define(TN_DONT, 254).
recv_banner(Sock, SoFar, UsedOpts) when byte_size(SoFar) > 4096 ->
    gen_tcp:close(Sock),
    [{open, true}, {banner, SoFar},
     {ended_at, <<"toolong">>}, {used_rfc854, UsedOpts}];
recv_banner(Sock, SoFar, UsedOpts) ->
    receive
        {tcp, Sock, <<?IAC, Cmd, Opt, Rest/binary>>} ->
            UsedOpts1 = case Cmd of
                ?TN_WILL -> gen_tcp:send(Sock, <<?IAC, ?TN_WONT, Opt>>), true;
                ?TN_DO -> gen_tcp:send(Sock, <<?IAC, ?TN_DONT, Opt>>), true;
                _ -> UsedOpts
            end,
            self() ! {tcp, Sock, Rest},
            recv_banner(Sock, SoFar, UsedOpts1);
        {tcp, Sock, Data} ->
            recv_banner(Sock, <<SoFar/binary, Data/binary>>, UsedOpts);
        {tcp_closed, Sock} ->
            [{open, true}, {banner, SoFar},
             {ended_at, <<"close">>}, {used_rfc854, UsedOpts}]
    after 2000 ->
        gen_tcp:close(Sock),
        [{open, true}, {banner, SoFar},
         {ended_at, <<"time">>}, {used_rfc854, UsedOpts}]
    end.

bulker_loop() ->
    bulker_loop([]).

bulker_loop(Q) when (length(Q) > 48) ->
    bulker_submit(Q),
    bulker_loop([]);
bulker_loop(Q) ->
    receive
        {add_data, Pid, Ip, Json} ->
            Pid ! {bulker, ok},
            bulker_loop([{Ip, Json} | Q])
    after 5000 ->
        bulker_submit(Q), bulker_loop([])
    end.

bulker_submit(Q) ->
    Uri = "http://localhost:9200/scanny/ip/_bulk",
    Data = iolist_to_binary([
        [to_json([{index, [{'_id', list_to_binary(ip_tools:to_str(Ip))}]}]),
            $\n, Json, $\n]
        || {Ip, Json} <- Q]),
    Headers = [{<<"Content-Type">>, <<"application/json">>}],
    case ssh_http:post(Uri, Headers, Data) of
        {ok, Code} when (Code >= 200) and (Code < 300) -> ok;
        {ok, Code} -> io:format("warning: got response code ~w: '~s'\n", [Code, Data]), ok;
        {error, _} -> bulker_submit(Q)
    end.

thread([], Acc) -> Acc;
thread([F | Rest], Acc) ->
    thread(Rest, F(Acc)).

binjoin([Next], _Sep) -> Next;
binjoin([Next | Rest], Sep) ->
    RestBin = binjoin(Rest, Sep),
    <<Next/binary, Sep/binary, RestBin/binary>>.

ts_to_jstime(TS) ->
    {{Year,Month,Day},{Hour,Min,Sec}} = calendar:now_to_universal_time(TS),
    lists:flatten(io_lib:format("~4..0w-~2..0w-~2..0wT~2..0w:~2..0w:~2..0wZ",
        [Year,Month,Day,Hour,Min,Sec])).

to_json(true) -> <<"true">>;
to_json(false) -> <<"false">>;
to_json(null) -> <<"null">>;
to_json({K, V}) when is_atom(K) ->
    KBin = atom_to_binary(K, utf8),
    VBin = to_json(V),
    <<"\"", KBin/binary, "\": ", VBin/binary>>;
to_json(Val) when is_list(Val) ->
    Inner = binjoin([to_json(V) || V <- Val], <<",">>),
    case Val of
        [{K,_} | _] when is_atom(K) -> <<"{", Inner/binary, "}">>;
        _ -> <<"[", Inner/binary, "]">>
    end;
to_json(Val) when is_binary(Val) ->
    Dirty = [<<I>> || I <- lists:seq(0,31)] ++ [<<I>> || I <- lists:seq(127,255)],
    Clean = thread([
        fun(V) -> binary:replace(V, <<$">>, <<"\\\"">>, [global]) end,
        fun(V) -> binary:replace(V, <<"\r\n">>, <<"  \\r\\n ">>, [global]) end,
        fun(V) -> binary:replace(V, <<"\n">>, <<" \\n ">>, [global]) end,
        fun(V) -> binary:replace(V, <<"\r">>, <<" \\r ">>, [global]) end,
        fun(V) -> binary:replace(V, Dirty, <<$?>>, [global]) end
    ], Val),
    <<"\"", Clean/binary, "\"">>;
to_json(Val) when is_integer(Val) ->
    integer_to_binary(Val).

pmap(Fun, GS, N) when is_tuple(GS) and is_integer(N) ->
    pmap(Fun, GS, N, []).
pmap(_Fun, {_G, done}, _N, []) -> ok;
pmap(Fun, GS = {_G, done}, N, Pids) ->
    receive {done, Pid} -> pmap(Fun, GS, N, Pids -- [Pid]) end;
pmap(Fun, GS, N, Pids) when length(Pids) >= N ->
    receive {done, Pid} -> pmap(Fun, GS, N, Pids -- [Pid]) end;
pmap(Fun, {GenFun, State}, N, Pids) ->
    Me = self(),
    Pid = spawn_link(fun() ->
        Fun(State),
        Me ! {done, self()}
    end),
    NextState = GenFun(State),
    pmap(Fun, {GenFun, NextState}, N, [Pid | Pids]).

ssh_http.erl

-module(ssh_http).

-define(DEFAULT_PACKET_SIZE, 32768).
-define(DEFAULT_WINDOW_SIZE, 2*?DEFAULT_PACKET_SIZE).
-define(DEFAULT_TIMEOUT, 5000).

-export([start_link/4, req/4, req/5]).
-export([get/2, get/3,
                post/3, post/4,
                put/3, put/4,
                delete/2, delete/3]).

-record(state, {ssh, mref, host, ssh_port, user, pw}).

start_link(Host, Port, User, Password) ->
        Pid = spawn_link(fun() -> ssh_master(Host, Port, User, Password) end),
        true = register(ssh_master, Pid),
        {ok, Pid}.

req(Method, Uri, Headers, Body) ->
        req(Method, Uri, Headers, Body, 5000).
req(Method, Uri, Headers, Body, Timeout) ->
        ssh_master ! {request, self(), Method, Uri, Headers, Body},
        receive
                {request_ok, Code} -> {ok, Code};
                {request_error, Err} -> {error, Err}
        after Timeout ->
                {error, timeout}
        end.

get(Uri, Headers) -> req(get, Uri, Headers, <<>>).
get(Uri, Headers, Timeout) -> req(get, Uri, Headers, <<>>, Timeout).
delete(Uri, Headers) -> req(delete, Uri, Headers, <<>>).
delete(Uri, Headers, Timeout) -> req(delete, Uri, Headers, <<>>, Timeout).
post(Uri, Headers, Body) -> req(post, Uri, Headers, Body).
post(Uri, Headers, Body, Timeout) -> req(post, Uri, Headers, Body, Timeout).
put(Uri, Headers, Body) -> req(put, Uri, Headers, Body).
put(Uri, Headers, Body, Timeout) -> req(put, Uri, Headers, Body, Timeout).

open_ssh(#state{host = Host, ssh_port = Port, user = User, pw = Pw}) ->
    ssh:connect(Host, Port, [
        {user, User}, {password, Pw},
        {quiet_mode, true}, {silently_accept_hosts, true}]).

open_tcp_chan(Host, Port, #state{ssh = Ssh}) when is_list(Host) ->
    HostBin = list_to_binary(Host), HostLen = byte_size(HostBin),
    OrigHost = <<"localhost">>, OrigHostLen = byte_size(OrigHost),
    OrigPort = crypto:rand_uniform(10000,65000),
    Msg = <<HostLen:32/big, HostBin/binary, Port:32/big, OrigHostLen:32/big, OrigHost/binary, OrigPort:32/big>>,
    ssh_connection_manager:open_channel(Ssh, "direct-tcpip", Msg, ?DEFAULT_WINDOW_SIZE, ?DEFAULT_PACKET_SIZE, ?DEFAULT_TIMEOUT).

ssh_master(Host, SshPort, User, Password) ->
        S = #state{host = Host, ssh_port = SshPort, user = User, pw = Password},
    {ok, Ssh} = open_ssh(S),
    SshMref = monitor(process, Ssh),
    chanreqs = ets:new(chanreqs, [set, public, named_table]),
    chans = ets:new(chans, [set, public, named_table]),
    ssh_master_loop(S#state{ssh = Ssh, mref = SshMref}).

ssh_master_loop(S = #state{ssh = Ssh, mref = SshMref}) ->
    Sz = ets:info(chanreqs, size),
    receive
        {'DOWN', SshMref, process, Ssh, Reason} ->
            io:format("lost ssh connection: ~999p\n", [Reason]),
            {ok, NewSsh} = open_ssh(S),
            NewSshMref = monitor(process, NewSsh),
            ets:delete_all_objects(chanreqs),
            ets:delete_all_objects(chans),
            ssh_master_loop(S#state{ssh = NewSsh, mref = NewSshMref});

        {'DOWN', _, process, ReqPid, Reason} ->
            case ets:lookup(chanreqs, ReqPid) of
                [{ReqPid, closed, _}] ->
                    ets:delete(chanreqs, ReqPid);
                [{ReqPid, none, CallerPid}] ->
                    CallerPid ! {request_error, Reason};
                [{ReqPid, Chan, CallerPid}] when (Reason =:= normal) ->
                    ets:delete(chans, Chan),
                    CallerPid ! {request_error, Reason};
                [{ReqPid, Chan, CallerPid}] ->
                    io:format("[~p] closed after crash\n", [Chan]),
                    ssh_connection:close(Ssh, Chan),
                    ets:delete(chans, Chan),
                    CallerPid ! {request_error, Reason};
                _ -> ok
            end,
            ssh_master_loop(S);

        {request, Pid, Method, Uri, Headers, Body} when (Sz < 16) ->
                {ok, {http, [], RemoteHost, RemotePort, PathStr, QueryStr}} = http_uri:parse(Uri),
                PathBin = list_to_binary(PathStr),
                QueryBin = list_to_binary(QueryStr),
                Path = <<PathBin/binary, QueryBin/binary>>,

            {Kid,_} = spawn_monitor(fun() ->
                receive go -> ok end,
                {ok, Chan} = open_tcp_chan(RemoteHost, RemotePort, S),
                true = ets:insert(chanreqs, {self(), Chan, Pid}),
                true = ets:insert(chans, {Chan, self()}),

                MethodBin = list_to_binary(string:to_upper(atom_to_list(Method))),
                Body0 = <<MethodBin/binary, " ", Path/binary, " HTTP/1.0\r\n">>,
                Body1 = lists:foldl(fun({K,V}, Acc) ->
                    <<Acc/binary, K/binary, ": ", V/binary, "\r\n">>
                end, Body0, Headers),
                Body2 = case byte_size(Body) of
                    0 -> Body1;
                    K -> B = integer_to_binary(K), <<Body1/binary, "Content-Length: ", B/binary, "\r\n">>
                end,
                Body3 = <<Body2/binary, "\r\n", Body/binary>>,
                ok = ssh_connection:send(Ssh, Chan, Body3),
                receive
                    {ssh_cm, Ssh, {data, Chan, _, Bin}} ->
                        [Head | _] = binary:split(Bin, <<"\r\n">>),
                        [_Ver, CodeBin | _Rest] = binary:split(Head, <<" ">>, [global]),
                        Code = binary_to_integer(CodeBin),
                        Pid ! {request_ok, Code}
                end,
                receive
                    {ssh_cm, Ssh, {closed, Chan}} ->
                        ets:insert(chanreqs, {self(), closed, Pid}),
                        ets:delete(chans, Chan)
                after 100 ->
                    ok
                end,
                ssh_connection:close(Ssh, Chan),
                ets:insert(chanreqs, {self(), closed, Pid}),
                ets:delete(chans, Chan)
            end),
            true = ets:insert(chanreqs, {Kid, none, Pid}),
            Kid ! go,
            ssh_master_loop(S)
    end.