add a genserver for handling remotes and start forwarding data

Current status is that p11tool successfully performs a --list-tokens
request over p11p, yay! Only once though -- the next request
hangs. heh.

Also, the timeout logic is wrong and should move to the server,
measuring the time it takes to get a full p11 response. This reqires
parsing of the p11-kit rpc protocol, which we need anyway.

4 files changed, 158 insertions, 45 deletions

diff --git a/p11p-daemon/src/p11p_remote.erl b/p11p-daemon/src/p11p_remote.erl
new file mode 100644
index 0000000..e89aa42
--- /dev/null
+++ b/p11p-daemon/src/p11p_remote.erl
@@ -0,0 +1,97 @@
+%% A remote spawns an Erlang port running the 'remote' program from
+%% p11-kit.
+
+%% Receive p11 requests from p11p_server, forward them to the remote,
+%% wait for a reply. If a reply is received within a timeout period,
+%% forward the reply to the requesting p11p_server. If the request
+%% times out, inform the remote manager (our parent).
+
+%% TODO: "remote" is not a great name and we shouldn't just inherit it
+%% from p11p-kit
+
+-module(p11p_remote).
+
+-behaviour(gen_server).
+
+%% API.
+-export([start_link/3]).
+-export([send/3]).
+
+%% Genserver callbacks.
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+         code_change/3]).
+
+%% Records and types.
+-record(state, {
+	  port :: port(),
+	  replyto :: pid() | undefined,
+	  timer :: reference() | undefined,
+	  token :: string()
+	 }).
+
+-define(P11KITREMOTE_PATH, "/home/linus/usr/libexec/p11-kit/p11-kit-remote").
+
+%% API.
+-spec start_link(atom(), string(), string()) -> {ok, pid()} | {error, term()}.
+start_link(ServName, TokName, ModPath) ->
+    lager:info("~p: p11p_remote starting for ~s", [ServName, ModPath]),
+    gen_server:start_link({local, ServName}, ?MODULE, [TokName, ModPath], []).
+
+-spec send(pid(), pid(), iodata()) -> ok.
+send(From, Remote, Data) ->
+    gen_server:cast(Remote, {send, From, Data}).
+
+%% Genserver callbacks.
+init([TokName, ModPath]) ->
+    Port = open_port({spawn_executable, ?P11KITREMOTE_PATH},
+		     [stream, exit_status, {args, [ModPath, "-v"]}]),
+    lager:debug("~s: New port: ~p", [?P11KITREMOTE_PATH, Port]),
+    {ok, #state{port = Port, token = TokName}}.
+
+handle_call(Request, _From, State) ->
+    lager:debug("Unhandled call: ~p~n", [Request]),
+    {reply, unhandled, State}.
+
+handle_cast({send, From, Data}, #state{port = Port} = State) ->
+    lager:debug("~p: sending ~B octets to remote ~p",
+		[self(), length(binary_to_list(Data)), Port]),
+    port_command(Port, Data),
+    %% TODO: move timing to server, measuring time for a p11 response
+    %% rather than this
+    Timer = erlang:start_timer(3000, self(), Port),
+    NewState = State#state{replyto = From, timer = Timer},
+    {noreply, NewState};
+handle_cast(Request, State) ->
+    lager:debug("Unhandled cast: ~p~n", [Request]),
+    {noreply, State}.
+
+handle_info({Port, {data, Data}}, #state{replyto = Pid, timer = Timer} = State) ->
+    if
+	Port == State#state.port ->
+	    erlang:cancel_timer(Timer, [{async, true}, {info, false}]),
+	    p11p_server:reply(Pid, Data);
+	true ->
+	    lager:debug("~p: data from unknown port ~p", [self(), Port])
+    end,
+    {noreply, State};
+handle_info({timeout, Timer, Port}, #state{token = TokName} = State) ->
+    NewState =
+	if
+	    Port == State#state.port andalso Timer == State#state.timer ->
+		p11p_remote_manager:timeout(TokName),
+		State#state{timer = undefined};
+	    true ->
+		lager:debug("~p: unknown timer ~p fired for port ~p",
+			    [self(), Timer, Port]),
+		State
+	end,
+    {noreply, NewState};
+handle_info(Info, State) ->
+    lager:debug("Unhandled info: ~p~n", [Info]),
+    {noreply, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_OldVersion, State, _Extra) ->
+    {ok, State}.
diff --git a/p11p-daemon/src/p11p_remote_manager.erl b/p11p-daemon/src/p11p_remote_manager.erl
index 281ae2b..3b3013b 100644
--- a/p11p-daemon/src/p11p_remote_manager.erl
+++ b/p11p-daemon/src/p11p_remote_manager.erl
@@ -12,8 +12,9 @@
 
 %% API.
 -export([start_link/0]).
--export([port_for_token/1]).			% For servers.
--export([p11init_done/1, timeout/0]).		% For remotes.
+-export([remote_for_token/1]).			% For servers.
+-export([p11init_done/1, timeout/1]).		% For remotes.
+-export([send/2]).				% Experiment
 
 %% Genserver callbacks.
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
@@ -22,35 +23,45 @@
 %% Records and types.
 -record(token, {
 	  p11init_done = false :: boolean(),
-	  remotes :: [port()]		      % Active remote in hd().
+	  remotes :: [pid()]		      % Active remote in hd().
 	 }).
 
 -record(state, {
 	  tokens :: #{string() => #token{}}
 	 }).
 
--define(P11KITREMOTE_PATH, "/home/linus/usr/libexec/p11-kit/p11-kit-remote").
-
 %% API implementation.
 -spec start_link() -> {ok, pid()} | {error, term()}.
 start_link() ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
-port_for_token(_Token) ->
-    todo.
+-spec remote_for_token(string()) -> pid().
+remote_for_token(TokName) ->
+    gen_server:call(?MODULE, {remote_for_token, TokName}).
 p11init_done(_Done) ->
     todo.
-timeout() ->
+timeout(_TokName) ->
     todo.
 
+send(TokName, Data) ->
+    gen_server:cast(?MODULE, {send, TokName, Data}).
+
 %% Genserver callbacks.
 init([]) ->
     {ok, #state{tokens = init_tokens(p11p_config:tokens())}}.
 
+handle_call({remote_for_token, TokName}, _From, State) ->
+    #{TokName := Token} = State#state.tokens,
+    {reply, hd(Token#token.remotes), State};
 handle_call(Request, _From, State) ->
     lager:debug("Unhandled call: ~p~n", [Request]),
     {reply, unhandled, State}.
 
+handle_cast({send, TokName, Data}, State) ->
+    #{TokName := Token} = State#state.tokens,
+    ServerPort = hd(Token#token.remotes),
+    ServerPort ! {self(), {command, Data}}, % TODO: Use synchronous port_command()?
+    {noreply, State};
 handle_cast(Request, State) ->
     lager:debug("Unhandled cast: ~p~n", [Request]),
     {noreply, State}.
@@ -59,7 +70,7 @@ handle_info({Port, {exit_status, Status}}, State) ->
     lager:info("~p: process exited with ~p", [Port, Status]),
     {stop, child_exit, State};
 handle_info(Info, State) ->
-    lager:debug("Unhandled info: ~p~n", [Info]),
+    lager:debug("~p: Unhandled info: ~p~n", [self(), Info]),
     {noreply, State}.
 
 terminate(_Reason, _State) ->
@@ -69,29 +80,27 @@ code_change(_OldVersion, State, _Extra) ->
     {ok, State}.
 
 %% Private functions
--spec init_tokens([any()]) -> #{string() => #token{}}.
+-spec init_tokens([p11p_config:token()]) -> #{string() => #token{}}.
 init_tokens(ConfTokens) ->
     init_tokens(ConfTokens, #{}).
 init_tokens([], Acc)->
     Acc;
 init_tokens([H|T], Acc)->
-    init_tokens(T, Acc#{p11p_config:nameof(H) => new_token(H)}).
+    TokName = p11p_config:nameof(H),
+    init_tokens(T, Acc#{TokName => new_token(TokName, H)}).
 
-new_token(ConfToken) ->
-    Remotes = start_remotes(p11p_config:modules_for_token(p11p_config:nameof(ConfToken))),
+new_token(TokName, ConfToken) ->
+    Remotes = start_remotes(TokName, p11p_config:modules_for_token(p11p_config:nameof(ConfToken))),
     #token{remotes = Remotes}.
 
-start_remotes(ConfModules) ->
-    start_remotes(ConfModules, []).
-start_remotes([], Acc) ->
-    lists:reverse(Acc);
-start_remotes([H|T], Acc) ->
+start_remotes(TokName, ConfModules) ->
+    start_remotes(TokName, ConfModules, []).
+start_remotes(_, [], Acc) ->
+    %%lists:reverse(Acc);
+    Acc;
+start_remotes(TokName, [H|T], Acc) ->
+    ModName = p11p_config:nameof(H),
+    ServName = "p11p_remote:" ++ TokName ++ ":" ++ ModName,
     ModPath = p11p_config:module_path(H),
-    Port = start_remote(ModPath),
-    start_remotes(T, [Port | Acc]).
-
-start_remote(ModPath) ->
-    Port = open_port({spawn_executable, ?P11KITREMOTE_PATH},
-		     [stream, exit_status, {args, [ModPath, "-v"]}]),
-    lager:debug("~s: New port: ~p", [?P11KITREMOTE_PATH, Port]),
-    Port.
+    {ok, Pid} = p11p_remote:start_link(list_to_atom(ServName), TokName, ModPath),
+    start_remotes(TokName, T, [Pid | Acc]).
diff --git a/p11p-daemon/src/p11p_server.erl b/p11p-daemon/src/p11p_server.erl
index 3ae5e78..4009c72 100644
--- a/p11p-daemon/src/p11p_server.erl
+++ b/p11p-daemon/src/p11p_server.erl
@@ -6,6 +6,7 @@
 
 %% API.
 -export([start_link/1]).
+-export([reply/2]).
 
 %% Genserver callbacks.
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
@@ -13,6 +14,7 @@
 
 %% Records and types.
 -record(state, {
+	  tokname :: string(),
 	  sockpath :: string(),		       % FIXME: filename(3erl)
 	  socket :: gen_tcp:socket()
 	 }).
@@ -20,28 +22,31 @@
 %% API.
 -spec start_link(gen_tcp:socket()) -> {ok, pid()} | {error, term()}.
 start_link(Args) ->
-    lager:debug("~p: p11p_server:start_link enter", [self()]),
     gen_server:start_link(?MODULE, Args, []).
 
+reply(Pid, Data) ->
+    gen_server:cast(Pid, {response, Data}).
+
 %% Genserver callbacks.
-init([SocketPath, Socket]) ->
+init([Token, SocketPath, Socket]) ->
     lager:debug("~p: p11p_server:init: ~s", [self(), SocketPath]),
     process_flag(trap_exit, true),		% We want terminate().
     gen_server:cast(self(), accept), % Perform accept in gen-server loop.
-    {ok, #state{sockpath = SocketPath, socket = Socket}}.
+    {ok, #state{tokname = Token, sockpath = SocketPath, socket = Socket}}.
 
 handle_call(Request, _From, State) ->
+
     lager:debug("Unhandled call: ~p~n", [Request]),
     {reply, unhandled, State}.
 
-handle_cast(accept, State = #state{sockpath = SocketPath, socket = ListenSocket}) ->
+handle_cast(accept, State = #state{tokname = TokName, sockpath = SocketPath, socket = ListenSocket}) ->
     %% Blocking until client connects or timeout fires. Without a
     %% timeout our supervisor cannot terminate us.
     case gen_tcp:accept(ListenSocket, 900) of
 	{ok, Sock} ->
 	    %% TODO: authz
 	    lager:debug("~p: ~p: new connection accepted", [self(), Sock]),
-	    p11p_server_sup:start_server([SocketPath, ListenSocket]), % Start a new acceptor.
+	    p11p_server_sup:start_server([TokName, SocketPath, ListenSocket]), % Start a new acceptor.
 	    {noreply, State#state{socket = Sock}}; % Use the new socket.
 	{error, timeout} ->
 	    gen_server:cast(self(), accept),	% Try again.
@@ -50,29 +55,31 @@ handle_cast(accept, State = #state{sockpath = SocketPath, socket = ListenSocket}
 	    lager:debug("~p: listening socket closed", [self()]),
 	    {stop, normal, State}
     end;
+handle_cast({response, Data}, #state{socket = ClientPort} = State) ->
+    lager:debug("~p: received ~B octets from remote", [self(), length(Data)]),
+    ok = gen_tcp:send(ClientPort, Data),
+    {noreply, State};
 handle_cast(Request, State) ->
     lager:debug("Unhandled cast: ~p~n", [Request]),
     {noreply, State}.
 
-handle_info({tcp, Port, Data}, State) ->
-    lager:debug("~p: received: ~s", [self(), Data]),
-    case Data of
-	<<"q\n">> ->
-	    gen_tcp:send(Port, "ok, bye\n"),	%DEBUG
-	    {stop, {shutdown, quit_by_client}, State};
-	_ ->
-	    gen_tcp:send(Port, "ok, thanks: " ++ Data),	%DEBUG
-	    {noreply, State}
-    end;
+handle_info({tcp, Port, Data}, #state{tokname = TokName} = State) ->
+    lager:debug("~p: received ~B octets from client on socket ~p",
+		[self(), length(binary_to_list(Data)), Port]),
+    %% TODO: parse incoming data, start timer per p11 request and
+    %% cancel after full response
+    Remote = p11p_remote_manager:remote_for_token(TokName),
+    ok = p11p_remote:send(self(), Remote, Data),
+    {noreply, State};
 handle_info({tcp_closed, Port}, State) ->
-    lager:debug("~p: ~p: closed", [self(), Port]),
+    lager:debug("~p: socket ~p closed", [self(), Port]),
     {stop, {shutdown, close_by_client}, State};
 handle_info(Info, State) ->
-    lager:debug("Unhandled info: ~p~n", [Info]),
+    lager:debug("~p: Unhandled info: ~p~n", [self(), Info]),
     {noreply, State}.
 
 terminate(_Reason, #state{sockpath = _SocketPath, socket = Socket}) ->
-    lager:debug("~p: ~p: terminated", [self(), Socket]),
+    lager:debug("~p: terminated", [self()]),
     gen_tcp:close(Socket),
     ok.
 
diff --git a/p11p-daemon/src/p11p_server_sup.erl b/p11p-daemon/src/p11p_server_sup.erl
index b38ea51..7ffa3d6 100644
--- a/p11p-daemon/src/p11p_server_sup.erl
+++ b/p11p-daemon/src/p11p_server_sup.erl
@@ -30,7 +30,7 @@ start_servers([Name|T]) ->
     file:delete(Path),
     % TODO: consider flow control through {active, once}, don't forget activating after read!
     {ok, Socket} = gen_tcp:listen(0, [{ifaddr, {local, Path}}, binary]),
-    spawn_link(?MODULE, start_server, [[Path, Socket]]),
+    spawn_link(?MODULE, start_server, [[Name, Path, Socket]]),
     start_servers(T).
 
 cleanup([]) ->