-module(lightbulb@services@access_token). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -export([fetch_access_token/3, set_authorization_header/2]). -export_type([access_token/0]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. -type access_token() :: {access_token, binary(), binary(), integer(), binary()}. -file("src/lightbulb/services/access_token.gleam", 110). -spec decode_access_token(binary()) -> {ok, access_token()} | {error, binary()}. decode_access_token(Body) -> Access_token_decoder = begin gleam@dynamic@decode:field( <<"access_token"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Token) -> gleam@dynamic@decode:field( <<"token_type"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Token_type) -> gleam@dynamic@decode:field( <<"expires_in"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_int/1}, fun(Expires_in) -> gleam@dynamic@decode:field( <<"scope"/utf8>>, {decoder, fun gleam@dynamic@decode:decode_string/1}, fun(Scope) -> gleam@dynamic@decode:success( {access_token, Token, Token_type, Expires_in, Scope} ) end ) end ) end ) end ) end, _pipe = gleam@json:parse(Body, Access_token_decoder), gleam@result:map_error( _pipe, fun(E) -> <<"Error decoding access token"/utf8, (gleam@string:inspect(E))/binary>> end ). -file("src/lightbulb/services/access_token.gleam", 63). -spec request_token( lightbulb@providers@http_provider:http_provider(), binary(), binary(), list(binary()) ) -> {ok, access_token()} | {error, binary()}. request_token(Http_provider, Url, Client_assertion, Scopes) -> Body = gleam@uri:query_to_string( [{<<"grant_type"/utf8>>, <<"client_credentials"/utf8>>}, {<<"client_assertion_type"/utf8>>, <<"urn:ietf:params:oauth:client-assertion-type:jwt-bearer"/utf8>>}, {<<"client_assertion"/utf8>>, Client_assertion}, {<<"scope"/utf8>>, gleam@string:join(Scopes, <<" "/utf8>>)}] ), gleam@result:'try'( begin _pipe = gleam@http@request:to(Url), gleam@result:replace_error( _pipe, <<"Error creating request for URL "/utf8, Url/binary>> ) end, fun(Req) -> Req@1 = begin _pipe@1 = Req, _pipe@2 = gleam@http@request:set_header( _pipe@1, <<"Content-Type"/utf8>>, <<"application/x-www-form-urlencoded"/utf8>> ), _pipe@3 = gleam@http@request:set_header( _pipe@2, <<"Accept"/utf8>>, <<"application/json"/utf8>> ), _pipe@4 = gleam@http@request:set_method(_pipe@3, post), gleam@http@request:set_body(_pipe@4, Body) end, case (erlang:element(2, Http_provider))(Req@1) of {ok, Resp} -> case erlang:element(2, Resp) of 200 -> decode_access_token(erlang:element(4, Resp)); 201 -> decode_access_token(erlang:element(4, Resp)); _ -> lightbulb@utils@logger:error_meta( <<"Error requesting access token"/utf8>>, Resp ), {error, <<"Error requesting access token"/utf8>>} end; E -> lightbulb@utils@logger:error_meta( <<"Error requesting access token"/utf8>>, E ), {error, <<"Error requesting access token"/utf8>>} end end ). -file("src/lightbulb/services/access_token.gleam", 171). -spec audience(binary(), gleam@option:option(binary())) -> binary(). audience(Auth_token_url, Auth_audience) -> case Auth_audience of none -> Auth_token_url; {some, <<""/utf8>>} -> Auth_token_url; {some, Audience} -> Audience end. -file("src/lightbulb/services/access_token.gleam", 131). -spec create_client_assertion( lightbulb@jwk:jwk(), binary(), binary(), gleam@option:option(binary()) ) -> binary(). create_client_assertion(Active_jwk, Auth_token_url, Client_id, Auth_audience) -> {_, Jwk} = begin _pipe = Active_jwk, lightbulb@jwk:to_map(_pipe) end, _assert_subject = ids@uuid:generate_v4(), {ok, Jti} = 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/services/access_token"/utf8>>, function => <<"create_client_assertion"/utf8>>, line => 140}) end, Jwt = maps:from_list( [{<<"iss"/utf8>>, gleam_stdlib:identity(Client_id)}, {<<"aud"/utf8>>, gleam_stdlib:identity(audience(Auth_token_url, Auth_audience))}, {<<"sub"/utf8>>, gleam_stdlib:identity(Client_id)}, {<<"iat"/utf8>>, begin _pipe@1 = birl:now(), _pipe@2 = birl:to_unix(_pipe@1), gleam_stdlib:identity(_pipe@2) end}, {<<"exp"/utf8>>, begin _pipe@3 = birl:now(), _pipe@4 = birl:add(_pipe@3, birl@duration:seconds(3600)), _pipe@5 = birl:to_unix(_pipe@4), gleam_stdlib:identity(_pipe@5) end}, {<<"jti"/utf8>>, gleam_stdlib:identity(Jti)}] ), Jws = maps:from_list( [{<<"alg"/utf8>>, <<"RS256"/utf8>>}, {<<"typ"/utf8>>, <<"JWT"/utf8>>}, {<<"kid"/utf8>>, erlang:element(2, Active_jwk)}] ), {_, Jose_jwt} = jose_jwt:sign(Jwk, Jws, Jwt), {_, Compact_signed} = jose_jws:compact(Jose_jwt), Compact_signed. -file("src/lightbulb/services/access_token.gleam", 39). ?DOC( " Requests an OAuth2 access token. Returns Ok(AccessToken) on success, Error(_) otherwise.\n" "\n" " As parameters, expects:\n" " 1. `providers`: A `Providers` instance that contains the HTTP provider and data provider.\n" " 2. `registration`: A `Registration` instance used to fetch the access token endpoint and client ID.\n" " 3. `scopes`: A list of scopes to request for the access token.\n" "\n" " Examples:\n" "\n" " ```gleam\n" " fetch_access_token(providers, registration, [\"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem\"])\n" " // Ok(AccessToken(\"actual_access_token\", \"Bearer\", 3600, \"https://purl.imsglobal.org/spec/lti-ags/scope/lineitem\"))\n" " ```\n" ). -spec fetch_access_token( lightbulb@providers:providers(), lightbulb@registration:registration(), list(binary()) ) -> {ok, access_token()} | {error, binary()}. fetch_access_token(Providers, Registration, Scopes) -> gleam@result:'try'( (erlang:element(6, erlang:element(2, Providers)))(), fun(Active_jwk) -> Client_assertion = create_client_assertion( Active_jwk, erlang:element(6, Registration), erlang:element(4, Registration), {some, erlang:element(6, Registration)} ), request_token( erlang:element(3, Providers), erlang:element(6, Registration), Client_assertion, Scopes ) end ). -file("src/lightbulb/services/access_token.gleam", 180). ?DOC(" Sets the Authorization header for a request using the provided access token.\n"). -spec set_authorization_header( gleam@http@request:request(binary()), access_token() ) -> gleam@http@request:request(binary()). set_authorization_header(Req, Access_token) -> {access_token, Token, _, _, _} = Access_token, _pipe = Req, gleam@http@request:set_header( _pipe, <<"Authorization"/utf8>>, <<"Bearer "/utf8, Token/binary>> ).