diff options
| -rw-r--r-- | merge/src/merge_sth.erl | 172 |
1 files changed, 172 insertions, 0 deletions
diff --git a/merge/src/merge_sth.erl b/merge/src/merge_sth.erl new file mode 100644 index 0000000..13d5204 --- /dev/null +++ b/merge/src/merge_sth.erl @@ -0,0 +1,172 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +-module(merge_sth). +-behaviour(gen_server). + +-export([start_link/1]). +-export([init/1, handle_call/3, terminate/2, handle_cast/2, handle_info/2, + code_change/3]). + +-include("plop.hrl"). + +-record(state, { + timer :: reference() + }). + +start_link(Args) -> + gen_server:start_link(?MODULE, Args, []). + +init([]) -> + lager:info("~p: starting", [?MODULE]), + {ok, #state{timer = erlang:start_timer(1000, self(), make_sth)}}. + +handle_call(stop, _From, State) -> + {stop, normal, stopped, State}. +handle_cast(_Request, State) -> + {noreply, State}. + +handle_info({timeout, _Timer, make_sth}, State) -> + make_sth(read_sth_treesize(), State). + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +terminate(Reason, #state{timer = Timer}) -> + lager:info("~p terminating: ~p", [?MODULE, Reason]), + erlang:cancel_timer(Timer), + ok. + +%%%%%%%%%%%%%%%%%%%% + +make_sth(noentry, State) -> + lager:info("Waiting for sth file."), + {noreply, State#state{timer = erlang:start_timer(1000, self(), make_sth)}}; +make_sth(CurSize, State) -> + {MergeSecondaryNames, _MergeSecondaryAddrs} = + lists:unzip(plopconfig:get_env(merge_secondaries, [])), + lager:info("Current STH at ~B with ~B secondary merge nodes.", + [CurSize, length(MergeSecondaryNames)]), + + %% Collect tree sizes from verified.* files in a list, add an + %% entry with the size found in the 'fetched' file, sort the list + %% (highest tree size first) and index it with backup quorum to + %% get our new tree size. + Sizes = [fetched() | verified_sizes(MergeSecondaryNames)], + BackupQuorumSize = plopconfig:get_env(backup_quorum, 0), + true = BackupQuorumSize =< length(MergeSecondaryNames), + NewSize = lists:nth(BackupQuorumSize + 1, Sizes), + lager:debug("new size at backup quorum ~B: ~B", [BackupQuorumSize, NewSize]), + + Wait = + if + NewSize < CurSize -> + lager:info("Waiting for enough backups to reach ~B, now at ~B.", + [CurSize, NewSize]), + 1; + true -> + ok = do_make_sth(NewSize), + max(1, round(application:get_env(plop, merge_delay, 600) / 60)) + end, + + Timer = erlang:start_timer(Wait * 1000, self(), make_sth), + {noreply, State#state{timer = Timer}}. + +do_make_sth(Size) -> + %% Build a new sth file in memory, get a signature for it and + %% verify both the new sth file against the signature and the new + %% root against ht before writing to disk. + NewTimestamp = plop:generate_timestamp(), + NewRoot = ht:root(Size - 1), + PackedSignature = make_signature(NewTimestamp, Size, NewRoot), + ok = case plop:verify_sth(Size, NewTimestamp, NewRoot, PackedSignature) of + true -> + NewSTH = [{"tree_size", Size}, + {"timestamp", NewTimestamp}, + {"sha256_root_hash", base64:encode(NewRoot)}, + {"tree_head_signature", base64:encode(PackedSignature)}], + ok = plop:save_sth({struct, NewSTH}), + ok; + false -> + lager:error("The signature we got for new tree of size ~B doesn't " ++ + "verify corectly; timestamp=~p; tree head: ~p", + [Size, NewTimestamp, NewRoot]), + sig_mismatch + end. + +make_signature(Timestamp, Size, Roothash) -> + SigType = plop:signature_type(tree_hash), + Sig = sign:sign_sth( + <<0:8, % CT protocol version (v1). + SigType:8, + Timestamp:64, + Size:64, + Roothash/binary>>), + plop:serialise(#signature{ + algorithm = #sig_and_hash_alg{ + hash_alg = sha256, + signature_alg = ecdsa}, + signature = Sig}). + +verified_sizes(MergeSecondaryNodeNames) -> + {ok, BasePath} = application:get_env(plop, verified_path), + L = lists:map(fun(NodeName) -> + verified_size(BasePath ++ "." ++ NodeName) + end, MergeSecondaryNodeNames), + lists:reverse(lists:sort(L)). + +verified_size(Path) -> + case atomic:readfile(Path) of + noentry -> + 0; + Contents -> + case mochijson2:decode(Contents) of + {struct, PropList} -> + Size = proplists:get_value(<<"tree_size">>, PropList), + Hash = hex:hexstr_to_bin(binary_to_list(proplists:get_value(<<"sha256_root_hash">>, PropList))), + ok = validate_tree_head(Size, Hash), + Size + end + end. + +validate_tree_head(Treesize, Roothash) -> + ht:load_tree(Treesize - 1), + ok = case ht:root(Treesize - 1) of + Roothash -> + ok; + RoothashInTree -> + lager:error("Root hash doesn't match tree head version ~B: ~p != ~p", [Treesize - 1, Roothash, RoothashInTree]), + root_mismatch + end. + +fetched() -> + case read_fetched() of + {-1, _} -> + 0; + {Index, Hash} -> + ok = merge_util:verify_logorder_and_fetched_consistency(Index, Hash), + Index + 1 + end. + +read_fetched() -> % FIXME: merge with backup implementation + case merge_util:readfile(fetched_path) of + noentry -> + {-1, <<>>}; + {struct, PropList} -> + {proplists:get_value(<<"index">>, PropList), + proplists:get_value(<<"hash">>, PropList)} + end. + + +read_sth_treesize() -> + case plop:sth() of + noentry -> + noentry; + {struct, STH} -> + Treesize = proplists:get_value(<<"tree_size">>, STH), + Timestamp = proplists:get_value(<<"timestamp">>, STH), + RootHash = base64:decode(proplists:get_value(<<"sha256_root_hash">>, STH)), + Signature = base64:decode(proplists:get_value(<<"tree_head_signature">>, STH)), + true = plop:verify_sth(Treesize, Timestamp, RootHash, Signature), + Treesize + end. |