<commit_before>%% ===================================================================
%% @author <NAME>
%% @copyright (C): 2013 ACK CYFRONET AGH
%% This software is released under the MIT license
%% cited in 'LICENSE.txt'.
%% @end
%% ===================================================================
%% @doc: This module forwards requests from socket to dispatcher.
%% @end
%% ===================================================================

%% TODO
%% Zrobić tak, żeby odpowiadając atomem nie trzeba było wysyłać do handlera
%% rekordu atom tylko, żeby handler sam przepakował ten atom do rekordu

-module(ws_handler).
-include("registered_names.hrl").
-include("messages_white_list.hrl").
-include("communication_protocol_pb.hrl").
-include("fuse_messages_pb.hrl").
-include("cluster_elements/request_dispatcher/gsi_handler.hrl").
-include("veil_modules/fslogic/fslogic.hrl").
-include("veil_modules/cluster_rengine/cluster_rengine.hrl").
-include("logging.hrl").
-include("veil_modules/dao/dao.hrl").
-include_lib("public_key/include/public_key.hrl").

%% Holds state of websocket connection. peer_dn field contains DN of certificate of connected peer.
-record(hander_state, {peer_dn, peer_serial, dispatcher_timeout, fuse_id = "", connection_id = ""}).

%% ====================================================================
%% API
%% ====================================================================
-export([init/3]).
-export([websocket_init/3]).
-export([websocket_handle/3]).
-export([websocket_info/3]).
-export([websocket_terminate/3]).

-ifdef(TEST).
-export([decode_protocol_buffer/2, encode_answer/2, encode_answer/3, encode_answer/5, encode_answer/6, checkMessage/2]).
-endif.

%% ====================================================================
%% API functions
%% ====================================================================

%% init/3
%% ====================================================================
%% @doc Switches protocol to WebSocket
-spec init(Proto :: term(), Req :: term(), Opts :: term()) -> {upgrade, protocol, cowboy_websocket}.
%% ====================================================================
init(_Proto, _Req, _Opts) ->
    {upgrade, protocol, cowboy_websocket}.


%% websocket_init/3
%% ====================================================================
%% @doc Cowboy's webscoket_init callback. Initialize connection, proceed with TLS-GSI authentication. <br/>
%%      If GSI validation fails, connection will be closed. <br/>
%%      Currently validation is handled by Globus NIF library loaded on erlang slave nodes.
-spec websocket_init(TransportName :: atom(), Req :: term(), Opts :: list()) -> {ok, Req :: term(), State :: term()} | {shutdown, Req :: term()}.
%% ====================================================================
websocket_init(TransportName, Req, _Opts) ->
    ?debug("WebSocket connection received. Transport: ~p", [TransportName]),
    {ok, PeerCert} = ssl:peercert(cowboy_req:get(socket, Req)),
    {ok, {Serial, Issuer}} = public_key:pkix_issuer_id(PeerCert, self),
    {ok, DispatcherTimeout} = application:get_env(veil_cluster_node, dispatcher_timeout),

    case ets:lookup(gsi_state, {Serial, Issuer}) of
        [{_, [OtpCert | Certs], _}] ->
            case gsi_handler:call(gsi_nif, verify_cert_c,
                [public_key:pkix_encode('OTPCertificate', OtpCert, otp),                    %% peer certificate
                    [public_key:pkix_encode('OTPCertificate', Cert, otp) || Cert <- Certs], %% peer CA chain
                    [DER || [DER] <- ets:match(gsi_state, {{ca, '_'}, '$1', '_'})],         %% cluster CA store
                    [DER || [DER] <- ets:match(gsi_state, {{crl, '_'}, '$1', '_'})]]) of    %% cluster CRL store
                {ok, 1} ->
                    {ok, EEC} = gsi_handler:find_eec_cert(OtpCert, Certs, gsi_handler:is_proxy_certificate(OtpCert)),
                    ?debug("Peer connected using certificate with subject: ~p ~n", [gsi_handler:proxy_subject(EEC)]),
                    {rdnSequence, Rdn} = gsi_handler:proxy_subject(EEC),
                    {ok, DnString} = user_logic:rdn_sequence_to_dn_string(Rdn),
                    {ok, Req, #hander_state{peer_dn = DnString, peer_serial = Serial, dispatcher_timeout = DispatcherTimeout}};
                {ok, 0, Errno} ->
                    ?info("Peer ~p was rejected due to ~p error code", [OtpCert#'OTPCertificate'.tbsCertificate#'OTPTBSCertificate'.subject, Errno]),
                    {shutdown, Req};
                {error, Reason} ->
                    ?error("GSI peer verification callback error: ~p", [Reason]),
                    {shutdown, Req};
                Other ->
                    ?error("GSI verification callback returned unknown response ~p", [Other]),
                    {shutdown, Req}
            end;
        _ ->
            ?error("Peer was conected but cerificate chain was not found. Please check if GSI validation is enabled."),
            {shutdown, Req}
    end.

%% websocket_handle/3
%% ====================================================================
%% @doc Cowboy's webscoket_handle callback. Binary data was received on socket. <br/>
%%      For more information please refer Cowboy's user manual.
-spec websocket_handle({Type :: atom(), Data :: term()}, Req, State) ->
    {reply, {Type :: atom(), Data :: term()}, Req, State} | {ok, Req, State} | {shutdown, Req, State}
    when
    Req :: term(),
    State :: #hander_state{}.
%% ====================================================================
websocket_handle({binary, Data}, Req, #hander_state{peer_dn = DnString} = State) ->
    try
        handle(Req, decode_protocol_buffer(Data, DnString), State) %% Decode ClusterMsg and handle it
    catch
        wrong_message_format                            -> {reply, {binary, encode_answer(wrong_message_format)}, Req, State};
        {wrong_internal_message_type, MsgId2}           -> {reply, {binary, encode_answer(wrong_internal_message_type, MsgId2)}, Req, State};
        {message_not_supported, MsgId2}                 -> {reply, {binary, encode_answer(message_not_supported, MsgId2)}, Req, State};
        {handshake_error, _HError, MsgId2}              -> {reply, {binary, encode_answer(handshake_error, MsgId2)}, Req, State};
        {no_user_found_error, _HError, MsgId2}          -> {reply, {binary, encode_answer(no_user_found_error, MsgId2)}, Req, State};
        {cert_confirmation_required, UserLogin, MsgId2} -> {reply, {binary, encode_answer(cert_confirmation_required, MsgId2, UserLogin)}, Req, State};
        {cert_denied_by_user, MsgId2}                   -> {reply, {binary, encode_answer(cert_denied_by_user, MsgId2)}, Req, State};
        {AtomError, MsgId2} when is_atom(AtomError)     -> {reply, {binary, encode_answer(AtomError, MsgId2)}, Req, State};
        _:_ -> {reply, {binary, encode_answer(ws_handler_error)}, Req, State}
    end;
websocket_handle({Type, Data}, Req, State) ->
    ?warning("Unknown WebSocket request. Type: ~p, Payload: ~p", [Type, Data]),
    {ok, Req, State}.

%% Internal websocket_handle method implementation
%% Handle Handshake request - FUSE ID negotiation
handle(Req, {_, _, Answer_decoder_name, ProtocolVersion,
    #handshakerequest{hostname = Hostname, variable = Vars, cert_confirmation = CertConfirmation} = HReq, MsgId, Answer_type},
    #hander_state{peer_dn = DnString} = State) ->
    ?debug("Handshake request: ~p", [HReq]),
    NewFuseId = genFuseId(HReq),
    UID = %% Fetch user's ID
    case user_logic:get_user({dn, DnString}) of
        {ok, #veil_document{uuid = UID1}} ->
            UID1;
        {error, Error} ->
            case user_logic:get_user({unverified_dn, DnString}) of
                {ok, #veil_document{uuid = UID1, record = #user{login = Login}} = UserDoc} ->
                    case CertConfirmation of
                        #handshakerequest_certconfirmation{login = Login, result = Result} ->
                            % Remove the DN from unverified DNs as it has been confirmed or declined
                            {ok, UserDoc2} = user_logic:update_unverified_dn_list(UserDoc, user_logic:get_unverified_dn_list(UserDoc) -- [DnString]),
                            case Result of
                                false ->
                                    ?alert("User ~p denied having added a certificate with DN: ~p", [Login, DnString]),
                                    throw({cert_denied_by_user, MsgId});
                                true ->
                                    {ok, _} = user_logic:update_dn_list(UserDoc2, user_logic:get_dn_list(UserDoc2) ++ [DnString]),
                                    ?debug("User ~p confirmed a certificate with DN: ~p", [Login, DnString]),
                                    UID1
                            end;
                        _ ->
                            ?debug("Handshake request is missing confirmation of certificate with DN: ~p. Denying connection.", [DnString]),
                            throw({cert_confirmation_required, Login, MsgId})
                    end;<commit_msg>VFS-608, fix log message when cert is declined<commit_after>%% ===================================================================
%% @author <NAME>
%% @copyright (C): 2013 ACK CYFRONET AGH
%% This software is released under the MIT license
%% cited in 'LICENSE.txt'.
%% @end
%% ===================================================================
%% @doc: This module forwards requests from socket to dispatcher.
%% @end
%% ===================================================================

%% TODO
%% Zrobić tak, żeby odpowiadając atomem nie trzeba było wysyłać do handlera
%% rekordu atom tylko, żeby handler sam przepakował ten atom do rekordu

-module(ws_handler).
-include("registered_names.hrl").
-include("messages_white_list.hrl").
-include("communication_protocol_pb.hrl").
-include("fuse_messages_pb.hrl").
-include("cluster_elements/request_dispatcher/gsi_handler.hrl").
-include("veil_modules/fslogic/fslogic.hrl").
-include("veil_modules/cluster_rengine/cluster_rengine.hrl").
-include("logging.hrl").
-include("veil_modules/dao/dao.hrl").
-include_lib("public_key/include/public_key.hrl").

%% Holds state of websocket connection. peer_dn field contains DN of certificate of connected peer.
-record(hander_state, {peer_dn, peer_serial, dispatcher_timeout, fuse_id = "", connection_id = ""}).

%% ====================================================================
%% API
%% ====================================================================
-export([init/3]).
-export([websocket_init/3]).
-export([websocket_handle/3]).
-export([websocket_info/3]).
-export([websocket_terminate/3]).

-ifdef(TEST).
-export([decode_protocol_buffer/2, encode_answer/2, encode_answer/3, encode_answer/5, encode_answer/6, checkMessage/2]).
-endif.

%% ====================================================================
%% API functions
%% ====================================================================

%% init/3
%% ====================================================================
%% @doc Switches protocol to WebSocket
-spec init(Proto :: term(), Req :: term(), Opts :: term()) -> {upgrade, protocol, cowboy_websocket}.
%% ====================================================================
init(_Proto, _Req, _Opts) ->
    {upgrade, protocol, cowboy_websocket}.


%% websocket_init/3
%% ====================================================================
%% @doc Cowboy's webscoket_init callback. Initialize connection, proceed with TLS-GSI authentication. <br/>
%%      If GSI validation fails, connection will be closed. <br/>
%%      Currently validation is handled by Globus NIF library loaded on erlang slave nodes.
-spec websocket_init(TransportName :: atom(), Req :: term(), Opts :: list()) -> {ok, Req :: term(), State :: term()} | {shutdown, Req :: term()}.
%% ====================================================================
websocket_init(TransportName, Req, _Opts) ->
    ?debug("WebSocket connection received. Transport: ~p", [TransportName]),
    {ok, PeerCert} = ssl:peercert(cowboy_req:get(socket, Req)),
    {ok, {Serial, Issuer}} = public_key:pkix_issuer_id(PeerCert, self),
    {ok, DispatcherTimeout} = application:get_env(veil_cluster_node, dispatcher_timeout),

    case ets:lookup(gsi_state, {Serial, Issuer}) of
        [{_, [OtpCert | Certs], _}] ->
            case gsi_handler:call(gsi_nif, verify_cert_c,
                [public_key:pkix_encode('OTPCertificate', OtpCert, otp),                    %% peer certificate
                    [public_key:pkix_encode('OTPCertificate', Cert, otp) || Cert <- Certs], %% peer CA chain
                    [DER || [DER] <- ets:match(gsi_state, {{ca, '_'}, '$1', '_'})],         %% cluster CA store
                    [DER || [DER] <- ets:match(gsi_state, {{crl, '_'}, '$1', '_'})]]) of    %% cluster CRL store
                {ok, 1} ->
                    {ok, EEC} = gsi_handler:find_eec_cert(OtpCert, Certs, gsi_handler:is_proxy_certificate(OtpCert)),
                    ?debug("Peer connected using certificate with subject: ~p ~n", [gsi_handler:proxy_subject(EEC)]),
                    {rdnSequence, Rdn} = gsi_handler:proxy_subject(EEC),
                    {ok, DnString} = user_logic:rdn_sequence_to_dn_string(Rdn),
                    {ok, Req, #hander_state{peer_dn = DnString, peer_serial = Serial, dispatcher_timeout = DispatcherTimeout}};
                {ok, 0, Errno} ->
                    ?info("Peer ~p was rejected due to ~p error code", [OtpCert#'OTPCertificate'.tbsCertificate#'OTPTBSCertificate'.subject, Errno]),
                    {shutdown, Req};
                {error, Reason} ->
                    ?error("GSI peer verification callback error: ~p", [Reason]),
                    {shutdown, Req};
                Other ->
                    ?error("GSI verification callback returned unknown response ~p", [Other]),
                    {shutdown, Req}
            end;
        _ ->
            ?error("Peer was conected but cerificate chain was not found. Please check if GSI validation is enabled."),
            {shutdown, Req}
    end.

%% websocket_handle/3
%% ====================================================================
%% @doc Cowboy's webscoket_handle callback. Binary data was received on socket. <br/>
%%      For more information please refer Cowboy's user manual.
-spec websocket_handle({Type :: atom(), Data :: term()}, Req, State) ->
    {reply, {Type :: atom(), Data :: term()}, Req, State} | {ok, Req, State} | {shutdown, Req, State}
    when
    Req :: term(),
    State :: #hander_state{}.
%% ====================================================================
websocket_handle({binary, Data}, Req, #hander_state{peer_dn = DnString} = State) ->
    try
        handle(Req, decode_protocol_buffer(Data, DnString), State) %% Decode ClusterMsg and handle it
    catch
        wrong_message_format                            -> {reply, {binary, encode_answer(wrong_message_format)}, Req, State};
        {wrong_internal_message_type, MsgId2}           -> {reply, {binary, encode_answer(wrong_internal_message_type, MsgId2)}, Req, State};
        {message_not_supported, MsgId2}                 -> {reply, {binary, encode_answer(message_not_supported, MsgId2)}, Req, State};
        {handshake_error, _HError, MsgId2}              -> {reply, {binary, encode_answer(handshake_error, MsgId2)}, Req, State};
        {no_user_found_error, _HError, MsgId2}          -> {reply, {binary, encode_answer(no_user_found_error, MsgId2)}, Req, State};
        {cert_confirmation_required, UserLogin, MsgId2} -> {reply, {binary, encode_answer(cert_confirmation_required, MsgId2, UserLogin)}, Req, State};
        {cert_denied_by_user, MsgId2}                   -> {reply, {binary, encode_answer(cert_denied_by_user, MsgId2)}, Req, State};
        {AtomError, MsgId2} when is_atom(AtomError)     -> {reply, {binary, encode_answer(AtomError, MsgId2)}, Req, State};
        _:_ -> {reply, {binary, encode_answer(ws_handler_error)}, Req, State}
    end;
websocket_handle({Type, Data}, Req, State) ->
    ?warning("Unknown WebSocket request. Type: ~p, Payload: ~p", [Type, Data]),
    {ok, Req, State}.

%% Internal websocket_handle method implementation
%% Handle Handshake request - FUSE ID negotiation
handle(Req, {_, _, Answer_decoder_name, ProtocolVersion,
    #handshakerequest{hostname = Hostname, variable = Vars, cert_confirmation = CertConfirmation} = HReq, MsgId, Answer_type},
    #hander_state{peer_dn = DnString} = State) ->
    ?debug("Handshake request: ~p", [HReq]),
    NewFuseId = genFuseId(HReq),
    UID = %% Fetch user's ID
    case user_logic:get_user({dn, DnString}) of
        {ok, #veil_document{uuid = UID1}} ->
            UID1;
        {error, Error} ->
            case user_logic:get_user({unverified_dn, DnString}) of
                {ok, #veil_document{uuid = UID1, record = #user{login = Login}} = UserDoc} ->
                    case CertConfirmation of
                        #handshakerequest_certconfirmation{login = Login, result = Result} ->
                            % Remove the DN from unverified DNs as it has been confirmed or declined
                            {ok, UserDoc2} = user_logic:update_unverified_dn_list(UserDoc, user_logic:get_unverified_dn_list(UserDoc) -- [DnString]),
                            case Result of
                                false ->
                                    ?alert("Private key owner denied having added a certificate with DN: ~p (added by ~p)", [DnString, Login]),
                                    throw({cert_denied_by_user, MsgId});
                                true ->
                                    {ok, _} = user_logic:update_dn_list(UserDoc2, user_logic:get_dn_list(UserDoc2) ++ [DnString]),
                                    ?debug("User ~p confirmed a certificate with DN: ~p", [Login, DnString]),
                                    UID1
                            end;
                        _ ->
                            ?debug("Handshake request is missing confirmation of certificate with DN: ~p. Denying connection.", [DnString]),
                            throw({cert_confirmation_required, Login, MsgId})
                    end;
{'fraction_non_alphanumeric': 0.217768798121553, 'fraction_numerical': 0.004859935564899252, 'mean_word_length': 3.1331527871812233, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4260123', 'n_tokens_mistral': 5433, 'n_tokens_neox': 5121, 'n_words': 1224}