FHIR Oauth

Thanks and fixed.

BTW, now the new version of this project is in python :

import time
import os
import json

import iris

from FhirInteraction import Interaction, Strategy, OAuthInteraction

from google.oauth2 import id_token
from google.auth.transport import requests

import requests as rq

# The following is an example of a custom OAuthInteraction class thatclassCustomOAuthInteraction(OAuthInteraction):
    
    client_id = None
    last_time_verified = None
    time_interval = 5defclear_instance(self):
        self.token_string = None
        self.oauth_client = None
        self.base_url = None
        self.username = None
        self.token_obj = None
        self.scopes = None
        self.verify_search_results = Nonedefset_instance(self, token:str,oauth_client:str,base_url:str,username:str):

        self.clear_instance()

        ifnot token ornot oauth_client:
            # the token or oauth client is not set, skip the verificationreturn

        global_time = iris.gref('^FHIR.OAuth2.Time')
        if global_time[token[0:50]]:
            self.last_time_verified = global_time[token[0:50]]

        if self.last_time_verified and (time.time() - self.last_time_verified) < self.time_interval:
            # the token was verified less than 5 seconds ago, skip the verificationreturn

        self.token_string = token
        self.oauth_client = oauth_client
        self.base_url = base_url
        self.username = username

        # try to set the client idtry:
            # first get the var env GOOGLE_CLIENT_ID is not set then None
            self.client_id = os.environ.get('GOOGLE_CLIENT_ID')
            # if not set, then by the secret.json fileifnot self.client_id:
                with open(os.environ.get('ISC_OAUTH_SECRET_PATH'),encoding='utf-8') as f:
                    data = json.load(f)
                    self.client_id = data['web']['client_id']
        except FileNotFoundError:
            passtry:
            self.verify_token(token)
        except Exception as e:
            self.clear_instance()
            raise e
        # token is valid, set the last time verified to now
        global_time[token[0:50]]=time.time()

    defverify_token(self,token:str):# check if the token is an access token or an id tokenif token.startswith('ya29.'):
            self.verify_access_token(token)
        else:
            self.verify_id_token(token)

    defverify_access_token(self,token:str):# verify the access token is valid# get with a timeout of 5 seconds
        response = rq.get(f"https://www.googleapis.com/oauth2/v3/tokeninfo?access_token={token}",timeout=5)
        try:
            response.raise_for_status()
        except rq.exceptions.HTTPError as e:
            # the token is not validraise e

    defverify_id_token(self,token:str):# Verify the token and get the user info
        idinfo = id_token.verify_oauth2_token(token, requests.Request(), self.client_id)
        if idinfo['iss'] notin ['accounts.google.com', 'https://accounts.google.com']:
            raise ValueError('Wrong issuer.')

    defget_introspection(self)->dict:return {}
    
    defget_user_info(self,basic_auth_username:str,basic_auth_roles:str)->dict:return {"Username":basic_auth_username,"Roles":basic_auth_roles}
    
    defverify_resource_id_request(self,resource_type:str,resource_id:str,required_privilege:str):passdefverify_resource_content(self,resource_dict:dict,required_privilege:str,allow_shared_resource:bool):passdefverify_history_instance_response(self,resource_type:str,resource_dict:dict,required_privilege:str):passdefverify_delete_request(self,resource_type:str,resource_id:str,required_privilege:str):passdefverify_search_request(self,
                              resource_type:str,
                              compartment_resource_type:str,
                              compartment_resource_id:str,
                              parameters:'iris.HS.FHIRServer.API.Data.QueryParameters',
                              required_privilege:str):passdefverify_system_level_request(self):passclassCustomStrategy(Strategy):defon_get_capability_statement(self, capability_statement):# Example : del resources Account
        capability_statement['rest'][0]['resource'] = [resource for resource in capability_statement['rest'][0]['resource'] if resource['type'] != 'Account']
        return capability_statement

classCustomInteraction(Interaction):defon_before_request(self, fhir_service, fhir_request, body, timeout):#Extract the user and roles for this request#so consent can be evaluated.
        self.requesting_user = fhir_request.Username
        self.requesting_roles = fhir_request.Roles

    defon_after_request(self, fhir_service, fhir_request, fhir_response, body):#Clear the user and roles between requests.
        self.requesting_user = ""
        self.requesting_roles = ""defpost_process_read(self, fhir_object):#Evaluate consent based on the resource and user/roles.#Returning 0 indicates this resource shouldn't be displayed - a 404 Not Found#will be returned to the user.return self.consent(fhir_object['resourceType'],
                        self.requesting_user,
                        self.requesting_roles)

    defpost_process_search(self, rs, resource_type):#Iterate through each resource in the search set and evaluate#consent based on the resource and user/roles.#Each row marked as deleted and saved will be excluded from the Bundle.
        rs._SetIterator(0)
        while rs._Next():
            ifnot self.consent(rs.ResourceType,
                            self.requesting_user,
                            self.requesting_roles):
                #Mark the row as deleted and save it.
                rs.MarkAsDeleted()
                rs._SaveRow()

    defconsent(self, resource_type, user, roles):#Example consent logic - only allow users with the role '%All' to see#Observation resources.if resource_type == 'Observation':
            if'%All'in roles:
                returnTrueelse:
                returnFalseelse:
            returnTrue

Based on https://community.intersystems.com/post/iris-fhir-python-strategy

yes, it is, i fixed the tutorial, thanks