diff options
| -rw-r--r-- | Emakefile | 7 | ||||
| -rw-r--r-- | README | 3 | ||||
| -rw-r--r-- | merge/ebin/merge.app | 13 | ||||
| -rw-r--r-- | merge/src/merge_app.erl | 12 | ||||
| -rw-r--r-- | merge/src/merge_dist.erl | 198 | ||||
| -rw-r--r-- | merge/src/merge_dist_sup.erl | 31 | ||||
| -rw-r--r-- | merge/src/merge_sup.erl | 21 | ||||
| -rw-r--r-- | src/frontend.erl | 6 | ||||
| -rw-r--r-- | src/plop_httputil.erl | 4 |
9 files changed, 286 insertions, 9 deletions
@@ -1,6 +1,11 @@ %% erl -make (-*- erlang -*-) {["src/*"], [debug_info, - {i, "../"}, % For hackney. + {i, "../"}, % For hackney. {outdir, "ebin/"}, {parse_transform, lager_transform}]}. +{["merge/src/*"], + [debug_info, + {i, "../"}, % For hackney. + {outdir, "merge/ebin/"}, + {parse_transform, lager_transform}]}. @@ -1,9 +1,6 @@ plop is a public log based on a Merkle tree. It can be used for implementing Certificate Transparency (RFC 6962). -The first implementation is in Erlang. The only interface supported -initially is Erlang messages. - Requires Erlang/OTP 17 [erts-6.0] or later. Compile the application diff --git a/merge/ebin/merge.app b/merge/ebin/merge.app new file mode 100644 index 0000000..80bafb9 --- /dev/null +++ b/merge/ebin/merge.app @@ -0,0 +1,13 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +%%% Application resource file for merge. + +{application, merge, + [{description, "Plop merge"}, + {vsn, "0.10.0-dev"}, + {modules, [merge_app, merge_sup, merge_dist_sup, merge_dist]}, + {applications, [kernel, stdlib, lager, plop]}, + {registered, []}, + {mod, {merge_app, []}} + ]}. diff --git a/merge/src/merge_app.erl b/merge/src/merge_app.erl new file mode 100644 index 0000000..cd68d81 --- /dev/null +++ b/merge/src/merge_app.erl @@ -0,0 +1,12 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +-module(merge_app). +-behaviour(application). +-export([start/2, stop/1]). + +start(normal, Args) -> + merge_sup:start_link(Args). + +stop(_State) -> + ok. diff --git a/merge/src/merge_dist.erl b/merge/src/merge_dist.erl new file mode 100644 index 0000000..54aaa00 --- /dev/null +++ b/merge/src/merge_dist.erl @@ -0,0 +1,198 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +-module(merge_dist). +-behaviour(gen_server). + +-export([start_link/1]). +-export([init/1, handle_call/3, terminate/2, handle_cast/2, handle_info/2, + code_change/3]). + +-record(state, { + timer :: reference(), + node_address :: string(), + sth_timestamp :: non_neg_integer() + }). + +start_link(Args) -> + gen_server:start_link(?MODULE, Args, []). + +init(Node) -> + lager:info("~p:~p: starting", [?MODULE, Node]), + Timer = erlang:start_timer(1000, self(), dist), + {ok, #state{timer = Timer, + node_address = Node, + sth_timestamp = 0}}. + +handle_call(stop, _From, State) -> + {stop, normal, stopped, State}. +handle_cast(_Request, State) -> + {noreply, State}. + +handle_info({timeout, _Timer, dist}, + #state{node_address = NodeAddress, + sth_timestamp = PrevTimestamp} = State) -> + case plop:sth() of + noentry -> + Timer = erlang:start_timer(1000, self(), dist), + {noreply, State#state{timer = Timer}}; + {struct, PropList} = STH -> + Treesize = proplists:get_value(<<"tree_size">>, PropList), + Timestamp = proplists:get_value(<<"timestamp">>, PropList), + RootHash = base64:decode(proplists:get_value(<<"sha256_root_hash">>, PropList)), + Signature = base64:decode(proplists:get_value(<<"tree_head_signature">>, PropList)), + Wait = max(1, round(application:get_env(plop, merge_delay, 600) / 60)), + case Timestamp > PrevTimestamp of + true -> + true = plop:verify_sth(Treesize, Timestamp, RootHash, Signature), + ok = dist(NodeAddress, min(Treesize, index:indexsize(logorder))), + ok = publish_sth(NodeAddress, STH), + lager:info("~p: Published STH with size ~B and timestamp ~p.", [NodeAddress, Treesize, Timestamp]), + {noreply, State#state{timer = erlang:start_timer(Wait * 1000, self(), dist), + sth_timestamp = Timestamp}}; + false -> + lager:debug("~p: STH timestamp ~p <= ~p, waiting ~B seconds.", [NodeAddress, Timestamp, PrevTimestamp, Wait]), + {noreply, State#state{timer = erlang:start_timer(Wait * 1000, self(), dist)}} + end + end. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +terminate(Reason, #state{timer = Timer}) -> + lager:info("~p terminating: ~p", [?MODULE, Reason]), + erlang:cancel_timer(Timer), + ok. + +publish_sth(NodeAddress, STH) -> + DebugTag = "publish-sth", + URL = NodeAddress ++ "publish-sth", + Headers = [{"Content-Type", "text/json"}], + RequestBody = list_to_binary(mochijson2:encode(STH)), + case plop_httputil:request(DebugTag, URL, Headers, RequestBody) of + {success, {_, StatusCode, _}, _, Body} when StatusCode == 200 -> + case (catch mochijson2:decode(Body)) of + {error, E} -> + lager:error("json parse error: ~p", [E]), + {error, E}; + {struct, PropList} -> + case proplists:get_value(<<"result">>, PropList) of + <<"ok">> -> ok; + Error -> {error, Error} + end + end + end. + +dist(NodeAddress, Size) -> + %% TODO: Lots of things here might fail bc node is not reachable, + %% in which case crashing is ok but restarting should wait some, + %% so what now? + {ok, VerifiedSize} = frontend_get_verifiedsize(NodeAddress), + true = Size >= VerifiedSize, + dist(NodeAddress, VerifiedSize, Size - VerifiedSize). + +dist(_, _, 0) -> + ok; +dist(NodeAddress, Size, NTotal) -> + DistMaxWindow = application:get_env(plop, merge_dist_winsize, 1000), + N = min(DistMaxWindow, NTotal), + Hashes = index:getrange(logorder, Size, Size + N - 1), + ok = frontend_sendlog(NodeAddress, Size, Hashes), + ok = frontend_send_missing_entries(NodeAddress, Hashes), + {ok, NewSize} = frontend_verify_entries(NodeAddress, Size + N), + lager:info("~p: Done distributing ~B entries.", [NodeAddress, NewSize-Size]), + true = NTotal >= NewSize - Size, + dist(NodeAddress, NewSize, NTotal - (NewSize - Size)). + +frontend_sendlog(NodeAddress, Start, Hashes) -> + SendlogChunksize = application:get_env(plop, merge_dist_sendlog_chunksize, 1000), + frontend_sendlog_chunk(NodeAddress, Start, + lists:split(min(SendlogChunksize, length(Hashes)), Hashes), + SendlogChunksize). + +frontend_sendlog_chunk(_, _, {[], _}, _) -> + ok; +frontend_sendlog_chunk(NodeAddress, Start, {Chunk, Rest}, Chunksize) -> + ok = frontend_sendlog_request(NodeAddress, Start, Chunk), + frontend_sendlog_chunk(NodeAddress, Start + length(Chunk), + lists:split(min(Chunksize, length(Rest)), Rest), + Chunksize). + +frontend_sendlog_request(NodeAddress, Start, Hashes) -> + DebugTag = io_lib:format("frontend sendlog ~B:~B", [Start, length(Hashes)]), + URL = NodeAddress ++ "sendlog", + Headers = [{"Content-Type", "text/json"}], + EncodedHashes = [base64:encode(H) || H <- Hashes], + RequestBody = list_to_binary(mochijson2:encode({[{"start", Start}, + {"hashes", EncodedHashes}]})), + case plop_httputil:request(DebugTag, URL, Headers, RequestBody) of + {success, {_, StatusCode, _}, _, Body} when StatusCode == 200 -> + case (catch mochijson2:decode(Body)) of + {error, E} -> + lager:error("json parse error: ~p", [E]), + {error, E}; + {struct, PropList} -> + case proplists:get_value(<<"result">>, PropList) of + <<"ok">> -> ok; + Error -> {error, Error} + end + end + end. + +frontend_send_missing_entries(NodeAddress, Hashes) -> + SendentriesChunksize = application:get_env(plop, merge_dist_sendentries_chunksize, 100), + {ChunkOfHashes, RestOfHashes} = lists:split(min(SendentriesChunksize, length(Hashes)), Hashes), + frontend_send_entries_chunk(NodeAddress, {ChunkOfHashes, RestOfHashes}, SendentriesChunksize). + +frontend_send_entries_chunk(_, {[], _}, _) -> + ok; +frontend_send_entries_chunk(NodeAddress, {Chunk, Rest}, Chunksize) -> + HashesAndEntries = lists:zip(Chunk, [db:entry_for_leafhash(H) || H <- Chunk]), + ok = frontend_send_entries_request(NodeAddress, HashesAndEntries), + frontend_send_entries_chunk(NodeAddress, + lists:split(min(Chunksize, length(Rest)), Rest), + Chunksize). + +frontend_send_entries_request(NodeAddress, HashesAndEntries) -> + DebugTag = io_lib:format("sendentry ~B", [length(HashesAndEntries)]), + URL = NodeAddress ++ "sendentry", + Headers = [{"Content-Type", "text/json"}], + L = mochijson2:encode( + [[{"entry", base64:encode(E)}, {"treeleafhash", base64:encode(H)}] || + {H, E} <- HashesAndEntries]), + RequestBody = list_to_binary(L), + case plop_httputil:request(DebugTag, URL, Headers, RequestBody) of + {success, {_, StatusCode, _}, _, Body} when StatusCode == 200 -> + case (catch mochijson2:decode(Body)) of + {error, E} -> + lager:error("json parse error: ~p", [E]), + {error, E}; + {struct, PropList} -> + case proplists:get_value(<<"result">>, PropList) of + <<"ok">> -> ok; + Error -> {error, Error} + end + end + end. + +frontend_get_verifiedsize(NodeAddress) -> + frontend_verify_entries(NodeAddress, 0). + +frontend_verify_entries(NodeAddress, Size) -> + DebugTag = io_lib:format("verify-entries ~B", [Size]), + URL = NodeAddress ++ "verify-entries", + Headers = [{"Content-Type", "text/json"}], + RequestBody = list_to_binary(mochijson2:encode({[{"verify_to", Size}]})), + case plop_httputil:request(DebugTag, URL, Headers, RequestBody) of + {success, {_, StatusCode, _}, _, Body} when StatusCode == 200 -> + case (catch mochijson2:decode(Body)) of + {error, E} -> + lager:error("json parse error: ~p", [E]), + {error, E}; + {struct, PropList} -> + case proplists:get_value(<<"result">>, PropList) of + <<"ok">> -> {ok, proplists:get_value(<<"verified">>, PropList)}; + Error -> {error, Error} + end + end + end. diff --git a/merge/src/merge_dist_sup.erl b/merge/src/merge_dist_sup.erl new file mode 100644 index 0000000..a4f7237 --- /dev/null +++ b/merge/src/merge_dist_sup.erl @@ -0,0 +1,31 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +-module(merge_dist_sup). +-behaviour(supervisor). + +-export([start_link/1, init/1]). + +start_link([]) -> + {ok, Nodes} = application:get_env(plop, frontend_nodes), + lager:info("starting merge dist for frontend nodes: ~p", [Nodes]), + {ok, Pid} = supervisor:start_link({local, ?MODULE}, ?MODULE, []), + Children = + lists:map(fun(Node) -> + lager:debug("starting dist worker: ~p", [Node]), + {ok, Child} = + supervisor:start_child(?MODULE, [Node]), + Child + end, Nodes), + lager:debug("supervisor ~p started dist workers: ~p", [Pid, Children]), + {ok, Pid}. + +init(_Args) -> + {ok, + {{simple_one_for_one, 3, 10}, + [ + {ignored, + {merge_dist, start_link, []}, + permanent, 10000, worker, + [merge_dist]} + ]}}. diff --git a/merge/src/merge_sup.erl b/merge/src/merge_sup.erl new file mode 100644 index 0000000..124fb12 --- /dev/null +++ b/merge/src/merge_sup.erl @@ -0,0 +1,21 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +-module(merge_sup). +-behaviour(supervisor). + +-export([start_link/1, init/1]). + +start_link(_Args) -> + supervisor:start_link({local, ?MODULE}, ?MODULE, []). + +init([]) -> + {ok, LogorderPath} = application:get_env(plop, index_path), + {ok, + {{one_for_one, 3, 10}, + [ + {the_logorder, {index, start_link, [logorder, LogorderPath]}, + permanent, 10000, worker, [index]}, + {merge_dist_sup, {merge_dist_sup, start_link, [[]]}, + transient, infinity, supervisor, [merge_dist_sup]} + ]}}. diff --git a/src/frontend.erl b/src/frontend.erl index e4d8e40..8a593b4 100644 --- a/src/frontend.erl +++ b/src/frontend.erl @@ -33,7 +33,7 @@ request(post, ?APPURL_PLOP_FRONTEND, "sendentry", Input) -> request(post, ?APPURL_PLOP_FRONTEND, "sendlog", Input) -> case (catch mochijson2:decode(Input)) of {error, E} -> - html("sendentry: bad input:", E); + html("sendlog: bad input:", E); {struct, PropList} -> Start = proplists:get_value(<<"start">>, PropList), Hashes = lists:map(fun (S) -> base64:decode(S) end, proplists:get_value(<<"hashes">>, PropList)), @@ -142,7 +142,7 @@ request(post, ?APPURL_PLOP_MERGE, "sendentry", Input) -> request(post, ?APPURL_PLOP_MERGE, "sendlog", Input) -> case (catch mochijson2:decode(Input)) of {error, E} -> - html("sendentry: bad input:", E); + html("sendlog: bad input:", E); {struct, PropList} -> Start = proplists:get_value(<<"start">>, PropList), Hashes = lists:map(fun (S) -> base64:decode(S) end, proplists:get_value(<<"hashes">>, PropList)), @@ -152,7 +152,7 @@ request(post, ?APPURL_PLOP_MERGE, "sendlog", Input) -> request(post, ?APPURL_PLOP_MERGE, "verifyroot", Input) -> case (catch mochijson2:decode(Input)) of {error, E} -> - html("sendentry: bad input:", E); + html("verifyroot: bad input:", E); {struct, PropList} -> OldSize = db:verifiedsize(), Treesize = proplists:get_value(<<"tree_size">>, PropList), diff --git a/src/plop_httputil.erl b/src/plop_httputil.erl index af4a5d1..b4188e7 100644 --- a/src/plop_httputil.erl +++ b/src/plop_httputil.erl @@ -64,14 +64,14 @@ read_and_verify_cacertfile(Filename) -> [KeyPem] = public_key:pem_decode(PemBin), {'Certificate', Der, _} = KeyPem, CalculatedHash = crypto:hash(sha256, Der), - CorrectHash = application:get_env(catlfish, https_cacert_fingerprint, none), + CorrectHash = application:get_env(plop, https_cacert_fingerprint, none), CorrectHash = CalculatedHash, Der. request(DebugTag, URL, Headers, RequestBody) -> Starttime = os:timestamp(), ParsedURL = hackney_url:parse_url(URL), - CACertFile = application:get_env(catlfish, https_cacertfile, none), + CACertFile = application:get_env(plop, https_cacertfile, none), CACert = read_and_verify_cacertfile(CACertFile), #hackney_url{path = Path, host = Host} = ParsedURL, lager:debug("~s: sending http request to ~p", |