Current section
Files
Jump to
Current section
Files
src/lightbulb@tool.erl
-module(lightbulb@tool).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([oidc_login/2, validate_launch/3]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-file("src/lightbulb/tool.gleam", 77).
-spec validate_issuer_exists(gleam@dict:dict(binary(), binary())) -> {ok,
gleam@dict:dict(binary(), binary())} |
{error, binary()}.
validate_issuer_exists(Params) ->
case gleam_stdlib:map_get(Params, <<"iss"/utf8>>) of
{ok, _} ->
{ok, Params};
{error, _} ->
{error, <<"Missing issuer"/utf8>>}
end.
-file("src/lightbulb/tool.gleam", 86).
-spec validate_login_hint_exists(gleam@dict:dict(binary(), binary())) -> {ok,
binary()} |
{error, binary()}.
validate_login_hint_exists(Params) ->
case gleam_stdlib:map_get(Params, <<"login_hint"/utf8>>) of
{ok, Login_hint} ->
{ok, Login_hint};
{error, _} ->
{error, <<"Missing login hint"/utf8>>}
end.
-file("src/lightbulb/tool.gleam", 95).
-spec validate_registration(
lightbulb@providers@data_provider:data_provider(),
gleam@dict:dict(binary(), binary())
) -> {ok, lightbulb@registration:registration()} | {error, binary()}.
validate_registration(Provider, Params) ->
gleam@result:'try'(
begin
_pipe = gleam_stdlib:map_get(Params, <<"iss"/utf8>>),
gleam@result:replace_error(_pipe, <<"Missing issuer"/utf8>>)
end,
fun(Issuer) ->
gleam@result:'try'(
begin
_pipe@1 = gleam_stdlib:map_get(Params, <<"client_id"/utf8>>),
gleam@result:replace_error(
_pipe@1,
<<"Missing client_id"/utf8>>
)
end,
fun(Client_id) ->
(erlang:element(4, Provider))(Issuer, Client_id)
end
)
end
).
-file("src/lightbulb/tool.gleam", 109).
-spec validate_client_id_exists(gleam@dict:dict(binary(), binary())) -> {ok,
binary()} |
{error, binary()}.
validate_client_id_exists(Params) ->
case gleam_stdlib:map_get(Params, <<"client_id"/utf8>>) of
{ok, Client_id} ->
{ok, Client_id};
{error, _} ->
{error, <<"Missing client id"/utf8>>}
end.
-file("src/lightbulb/tool.gleam", 145).
-spec validate_oidc_state(gleam@dict:dict(binary(), LGM), LGM) -> {ok, LGM} |
{error, binary()}.
validate_oidc_state(Params, Session_state) ->
gleam@result:'try'(
begin
_pipe = gleam_stdlib:map_get(Params, <<"state"/utf8>>),
gleam@result:replace_error(_pipe, <<"Missing state"/utf8>>)
end,
fun(State) -> case State =:= Session_state of
true ->
{ok, State};
false ->
{error, <<"Invalid state"/utf8>>}
end end
).
-file("src/lightbulb/tool.gleam", 173).
-spec peek_claim(binary(), binary(), gleam@dynamic@decode:decoder(LDB)) -> {ok,
LDB} |
{error, binary()}.
peek_claim(Jwt_string, Claim, Decoder) ->
case jose_jwt:peek(Jwt_string) of
{jose_jwt, Claims} ->
case gleam_stdlib:map_get(Claims, Claim) of
{ok, Value} ->
_pipe = gleam@dynamic@decode:run(Value, Decoder),
gleam@result:replace_error(_pipe, <<"Invalid claim"/utf8>>);
{error, _} ->
{error, <<"Missing claim"/utf8>>}
end
end.
-file("src/lightbulb/tool.gleam", 166).
-spec peek_issuer_client_id(binary()) -> {ok, {binary(), binary()}} |
{error, binary()}.
peek_issuer_client_id(Id_token) ->
gleam@result:'try'(
peek_claim(
Id_token,
<<"iss"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Issuer) ->
gleam@result:'try'(
peek_claim(
Id_token,
<<"aud"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Client_id) -> {ok, {Issuer, Client_id}} end
)
end
).
-file("src/lightbulb/tool.gleam", 157).
-spec peek_validate_registration(
binary(),
lightbulb@providers@data_provider:data_provider()
) -> {ok, lightbulb@registration:registration()} | {error, binary()}.
peek_validate_registration(Id_token, Provider) ->
gleam@result:'try'(
peek_issuer_client_id(Id_token),
fun(_use0) ->
{Issuer, Client_id} = _use0,
(erlang:element(4, Provider))(Issuer, Client_id)
end
).
-file("src/lightbulb/tool.gleam", 185).
-spec peek_header_claim(binary(), binary(), gleam@dynamic@decode:decoder(LDF)) -> {ok,
LDF} |
{error, binary()}.
peek_header_claim(Jwt_string, Header, Decoder) ->
case jose_jwt:peek_protected(Jwt_string) of
{jose_jws, _, _, Headers} ->
case gleam_stdlib:map_get(Headers, Header) of
{ok, Value} ->
_pipe = gleam@dynamic@decode:run(Value, Decoder),
gleam@result:replace_error(_pipe, <<"Invalid header"/utf8>>);
{error, _} ->
{error, <<"Missing header"/utf8>>}
end
end.
-file("src/lightbulb/tool.gleam", 215).
-spec fetch_jwk(binary(), binary()) -> {ok, gleam@dict:dict(binary(), binary())} |
{error, binary()}.
fetch_jwk(Keyset_url, Kid) ->
gleam@result:'try'(
begin
_pipe = gleam@http@request:to(Keyset_url),
gleam@result:replace_error(_pipe, <<"Invalid keyset URL"/utf8>>)
end,
fun(Req) ->
Req@1 = gleam@http@request:prepend_header(
Req,
<<"accept"/utf8>>,
<<"application/vnd.hmrc.1.0+json"/utf8>>
),
gleam@result:'try'(
begin
_pipe@1 = gleam@httpc:send(Req@1),
gleam@result:replace_error(
_pipe@1,
<<"Failed to fetch keyset from "/utf8,
Keyset_url/binary>>
)
end,
fun(Resp) -> case Resp of
{response, 200, _, Body} ->
Keyset_decoder = begin
gleam@dynamic@decode:field(
<<"keys"/utf8>>,
gleam@dynamic@decode:list(
gleam@dynamic@decode:dict(
{decoder,
fun gleam@dynamic@decode:decode_string/1},
{decoder,
fun gleam@dynamic@decode:decode_string/1}
)
),
fun(Keys) ->
gleam@dynamic@decode:success(Keys)
end
)
end,
case gleam@json:parse(Body, Keyset_decoder) of
{ok, Keys@1} ->
_pipe@2 = gleam@list:find(
Keys@1,
fun(Key) ->
gleam_stdlib:map_get(
Key,
<<"kid"/utf8>>
)
=:= {ok, Kid}
end
),
gleam@result:map_error(
_pipe@2,
fun(_) ->
lightbulb@utils@logger:error_meta(
<<"Failed to find JWK with kid "/utf8,
Kid/binary>>,
Keys@1
),
<<"Failed to find JWK with kid"/utf8>>
end
);
{error, _} ->
lightbulb@utils@logger:error_meta(
<<"Failed to parse keyset"/utf8>>,
{Keyset_url, Body}
),
{error, <<"Failed to parse keyset"/utf8>>}
end;
{response, Status, _, _} ->
lightbulb@utils@logger:error_meta(
<<"Failed to fetch keyset"/utf8>>,
{Status, Keyset_url}
),
{error, <<"Failed to fetch keyset"/utf8>>}
end end
)
end
).
-file("src/lightbulb/tool.gleam", 197).
-spec verify_token(binary(), binary()) -> {ok,
gleam@dict:dict(binary(), gleam@dynamic:dynamic_())} |
{error, binary()}.
verify_token(Id_token, Keyset_url) ->
gleam@result:'try'(
begin
_pipe = peek_header_claim(
Id_token,
<<"kid"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
gleam@result:replace_error(_pipe, <<"Missing kid"/utf8>>)
end,
fun(Kid) ->
gleam@result:'try'(
fetch_jwk(Keyset_url, Kid),
fun(Jwk) ->
case jose_jwt:verify(jose_jwk:from_map(Jwk), Id_token) of
{true, {jose_jwt, Claims}, _} ->
{ok, Claims};
_ ->
lightbulb@utils@logger:error_meta(
<<"Failed to verify id_token"/utf8>>,
Id_token
),
{error, <<"Failed to verify id_token"/utf8>>}
end
end
)
end
).
-file("src/lightbulb/tool.gleam", 356).
-spec get_claim(
gleam@dict:dict(binary(), gleam@dynamic:dynamic_()),
binary(),
gleam@dynamic@decode:decoder(LDW)
) -> {ok, LDW} | {error, binary()}.
get_claim(Claims, Claim, Decoder) ->
_pipe = gleam_stdlib:map_get(Claims, Claim),
_pipe@2 = gleam@result:map(
_pipe,
fun(C) -> _pipe@1 = gleam@dynamic@decode:run(C, Decoder),
gleam@result:replace_error(
_pipe@1,
<<"Invalid claim "/utf8, Claim/binary>>
) end
),
_pipe@3 = gleam@result:replace_error(
_pipe@2,
<<"Missing claim "/utf8, Claim/binary>>
),
gleam@result:flatten(_pipe@3).
-file("src/lightbulb/tool.gleam", 281).
-spec validate_timestamps(gleam@dict:dict(binary(), gleam@dynamic:dynamic_())) -> {ok,
gleam@dict:dict(binary(), gleam@dynamic:dynamic_())} |
{error, binary()}.
validate_timestamps(Claims) ->
gleam@result:'try'(
begin
_pipe = get_claim(
Claims,
<<"exp"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1}
),
gleam@result:map(_pipe, fun birl:from_unix/1)
end,
fun(Exp) ->
gleam@result:'try'(
begin
_pipe@1 = get_claim(
Claims,
<<"iat"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_int/1}
),
gleam@result:map(_pipe@1, fun birl:from_unix/1)
end,
fun(Iat) ->
Now = birl:now(),
Buffer_sec = 2,
A_few_seconds_ago = birl:subtract(
Now,
birl@duration:seconds(Buffer_sec)
),
A_few_seconds_later = birl:add(
Now,
birl@duration:seconds(Buffer_sec)
),
gleam@bool:guard(
birl:compare(Exp, A_few_seconds_ago) =:= lt,
{error, <<"JWT exp is expired"/utf8>>},
fun() ->
gleam@bool:guard(
birl:compare(Iat, A_few_seconds_later) =:= gt,
{error, <<"JWT iat is in the future"/utf8>>},
fun() -> {ok, Claims} end
)
end
)
end
)
end
).
-file("src/lightbulb/tool.gleam", 307).
-spec validate_nonce(
gleam@dict:dict(binary(), gleam@dynamic:dynamic_()),
lightbulb@providers@data_provider:data_provider()
) -> {ok, gleam@dict:dict(binary(), gleam@dynamic:dynamic_())} |
{error, binary()}.
validate_nonce(Claims, Provider) ->
gleam@result:'try'(
get_claim(
Claims,
<<"nonce"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Nonce) -> case (erlang:element(3, Provider))(Nonce) of
{ok, _} ->
{ok, Claims};
{error, E} ->
lightbulb@utils@logger:error_meta(
<<"Failed to validate nonce: "/utf8, E/binary>>,
Claims
),
{error, E}
end end
).
-file("src/lightbulb/tool.gleam", 266).
-spec validate_deployment(
gleam@dict:dict(binary(), gleam@dynamic:dynamic_()),
lightbulb@providers@data_provider:data_provider(),
binary(),
binary()
) -> {ok, lightbulb@deployment:deployment()} | {error, binary()}.
validate_deployment(Claims, Provider, Issuer, Client_id) ->
gleam@result:'try'(
get_claim(
Claims,
<<"https://purl.imsglobal.org/spec/lti/claim/deployment_id"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Deployment_id) ->
(erlang:element(5, Provider))(Issuer, Client_id, Deployment_id)
end
).
-file("src/lightbulb/tool.gleam", 321).
-spec validate_lti_resource_link_request_message(
gleam@dict:dict(binary(), gleam@dynamic:dynamic_())
) -> {ok, gleam@dict:dict(binary(), gleam@dynamic:dynamic_())} |
{error, binary()}.
validate_lti_resource_link_request_message(Claims) ->
case get_claim(
Claims,
<<"https://purl.imsglobal.org/spec/lti/claim/message_type"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1}
) of
{ok, <<"LtiResourceLinkRequest"/utf8>>} ->
{ok, Claims};
_ ->
lightbulb@utils@logger:error_meta(
<<"Invalid message type"/utf8>>,
{Claims,
<<"https://purl.imsglobal.org/spec/lti/claim/message_type"/utf8>>}
),
{error, <<"Invalid message type"/utf8>>}
end.
-file("src/lightbulb/tool.gleam", 32).
?DOC(
" Builds an OIDC login response for the tool. This function will return a `state` and `redirect_url`.\n"
" The `state` is an opaque string that will be used to verify the response from the\n"
" OIDC provider. The `redirect_url` is the URL that the user will be redirected to\n"
" to authenticate and complete the OIDC login process.\n"
" (LTI 1.3 Specification)[https://www.imsglobal.org/spec/lti/v1p3#oidc-login-request] for more details.\n"
).
-spec oidc_login(
lightbulb@providers@data_provider:data_provider(),
gleam@dict:dict(binary(), binary())
) -> {ok, {binary(), binary()}} | {error, binary()}.
oidc_login(Provider, Params) ->
gleam@result:'try'(
validate_issuer_exists(Params),
fun(_) ->
gleam@result:'try'(
begin
_pipe = gleam_stdlib:map_get(
Params,
<<"target_link_uri"/utf8>>
),
gleam@result:replace_error(
_pipe,
<<"Missing target_link_uri"/utf8>>
)
end,
fun(Target_link_uri) ->
gleam@result:'try'(
validate_login_hint_exists(Params),
fun(Login_hint) ->
gleam@result:'try'(
validate_registration(Provider, Params),
fun(Registration) ->
gleam@result:'try'(
validate_client_id_exists(Params),
fun(Client_id) ->
_assert_subject = ids@uuid:generate_v4(
),
{ok, State} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(
#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"lightbulb/tool"/utf8>>,
function => <<"oidc_login"/utf8>>,
line => 45}
)
end,
_assert_subject@1 = (erlang:element(
2,
Provider
))(),
{ok, Nonce} = case _assert_subject@1 of
{ok, _} -> _assert_subject@1;
_assert_fail@1 ->
erlang:error(
#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail@1,
module => <<"lightbulb/tool"/utf8>>,
function => <<"oidc_login"/utf8>>,
line => 46}
)
end,
Query_params = [{<<"scope"/utf8>>,
<<"openid"/utf8>>},
{<<"response_type"/utf8>>,
<<"id_token"/utf8>>},
{<<"response_mode"/utf8>>,
<<"form_post"/utf8>>},
{<<"prompt"/utf8>>,
<<"none"/utf8>>},
{<<"client_id"/utf8>>,
Client_id},
{<<"redirect_uri"/utf8>>,
Target_link_uri},
{<<"state"/utf8>>, State},
{<<"nonce"/utf8>>,
erlang:element(2, Nonce)},
{<<"login_hint"/utf8>>,
Login_hint}],
Query_params@1 = case gleam_stdlib:map_get(
Params,
<<"lti_message_hint"/utf8>>
) of
{ok, Lti_message_hint} ->
[{<<"lti_message_hint"/utf8>>,
Lti_message_hint} |
Query_params];
{error, _} ->
Query_params
end,
Redirect_url = <<<<(erlang:element(
5,
Registration
))/binary,
"?"/utf8>>/binary,
(gleam@uri:query_to_string(
Query_params@1
))/binary>>,
{ok, {State, Redirect_url}}
end
)
end
)
end
)
end
)
end
).
-file("src/lightbulb/tool.gleam", 338).
-spec validate_message(gleam@dict:dict(binary(), gleam@dynamic:dynamic_())) -> {ok,
gleam@dict:dict(binary(), gleam@dynamic:dynamic_())} |
{error, binary()}.
validate_message(Claims) ->
gleam@result:'try'(
get_claim(
Claims,
<<"https://purl.imsglobal.org/spec/lti/claim/message_type"/utf8>>,
{decoder, fun gleam@dynamic@decode:decode_string/1}
),
fun(Message_type) ->
gleam@result:'try'(
begin
_pipe = gleam@list:find(
[{<<"LtiResourceLinkRequest"/utf8>>,
fun validate_lti_resource_link_request_message/1}],
fun(Validator) ->
erlang:element(1, Validator) =:= Message_type
end
),
gleam@result:replace_error(
_pipe,
<<"No validator found for message type "/utf8,
Message_type/binary>>
)
end,
fun(_use0) ->
{_, Validator@1} = _use0,
_pipe@1 = Validator@1(Claims),
gleam@result:replace_error(
_pipe@1,
<<"Invalid message type "/utf8, Message_type/binary>>
)
end
)
end
).
-file("src/lightbulb/tool.gleam", 121).
?DOC(
" Validates the OIDC login response from the OIDC provider. This function will validate and unpack\n"
" the `id_token` and return claims as a map if the token is valid. The `state` parametrer is the\n"
" opaque string that was stored in a cookie during `oidc_login` step.\n"
).
-spec validate_launch(
lightbulb@providers@data_provider:data_provider(),
gleam@dict:dict(binary(), binary()),
binary()
) -> {ok, gleam@dict:dict(binary(), gleam@dynamic:dynamic_())} |
{error, binary()}.
validate_launch(Provider, Params, Session_state) ->
gleam@result:'try'(
begin
_pipe = gleam_stdlib:map_get(Params, <<"id_token"/utf8>>),
gleam@result:replace_error(_pipe, <<"Missing id_token"/utf8>>)
end,
fun(Id_token) ->
gleam@result:'try'(
validate_oidc_state(Params, Session_state),
fun(_) ->
gleam@result:'try'(
peek_validate_registration(Id_token, Provider),
fun(Registration) ->
gleam@result:'try'(
verify_token(
Id_token,
erlang:element(7, Registration)
),
fun(Claims) ->
gleam@result:'try'(
validate_deployment(
Claims,
Provider,
erlang:element(3, Registration),
erlang:element(4, Registration)
),
fun(_) ->
gleam@result:'try'(
validate_timestamps(Claims),
fun(_) ->
gleam@result:'try'(
validate_nonce(
Claims,
Provider
),
fun(_) ->
gleam@result:'try'(
validate_message(
Claims
),
fun(_) ->
{ok, Claims}
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end
).