title: Unable to trigger composer/airflow dag from Cloud function that triggers when there are changes in cloud storage
body: <p>I have created and ran dags on a <code>google-cloud-composer</code> environment (<strong>dlkpipelinesv1</strong> : composer-1.13.0-airflow-1.10.12). I am able to trigger these dags manually, and using the scheduler, but I am stuck when it comes to triggering them via <code>cloud-functions</code> that detect changes in a <code>google-cloud-storage</code> bucket.</p>
  <p>Note that I had another GC-Composer environment (<strong>pipelines</strong>:composer-1.7.5-airflow-1.10.2) that used those same google cloud functions to trigger the relevant dags, and <strong>it was working</strong>.</p>
  <p>I followed <a href="https://cloud.google.com/composer/docs/how-to/using/triggering-with-gcf" rel="nofollow noreferrer">this guide</a> to create the functions that trigger the dags. So I retrieved the following variables:</p>
  <pre><code>PROJECT_ID = &lt;project_id&gt;
  CLIENT_ID = &lt;client_id_retrieved_by_running_the_code_in_the_guide_within_my_gcp_console&gt;
  WEBSERVER_ID = &lt;airflow_webserver_id&gt;
  DAG_NAME = &lt;dag_to_trigger&gt;
  WEBSERVER_URL = f&quot;https://{WEBSERVER_ID}.appspot.com/api/experimental/dags/{DAG_NAME}/dag_runs&quot;
  
  
  def file_listener(event, context):
      &quot;&quot;&quot;Entry point of the cloud function: Triggered by a change to a Cloud Storage bucket.
      Args:
           event (dict): Event payload.
           context (google.cloud.functions.Context): Metadata for the event.
      &quot;&quot;&quot;
      logging.info(&quot;Running the file listener process&quot;)
      logging.info(f&quot;event : {event}&quot;)
      logging.info(f&quot;context : {context}&quot;)
      file = event
      if file[&quot;size&quot;] == &quot;0&quot; or &quot;DTM_DATALAKE_AUDIT_COMPTAGE&quot; not in file[&quot;name&quot;] or &quot;.filepart&quot; in file[&quot;name&quot;].lower():
          logging.info(&quot;no matching file&quot;)
          exit(0)
  
      logging.info(f&quot;File listener detected the presence of : {file['name']}.&quot;)
  
      # id_token = authorize_iap()
      # make_iap_request({&quot;file_name&quot;: file[&quot;name&quot;]}, id_token)
      make_iap_request(url=WEBSERVER_URL, client_id=CLIENT_ID, method=&quot;POST&quot;)
  
  
  def make_iap_request(url, client_id, method=&quot;GET&quot;, **kwargs):
      &quot;&quot;&quot;Makes a request to an application protected by Identity-Aware Proxy.
  
      Args:
        url: The Identity-Aware Proxy-protected URL to fetch.
        client_id: The client ID used by Identity-Aware Proxy.
        method: The request method to use
                ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE')
        **kwargs: Any of the parameters defined for the request function:
                  https://github.com/requests/requests/blob/master/requests/api.py
                  If no timeout is provided, it is set to 90 by default.
  
      Returns:
        The page body, or raises an exception if the page couldn't be retrieved.
      &quot;&quot;&quot;
      # Set the default timeout, if missing
      if &quot;timeout&quot; not in kwargs:
          kwargs[&quot;timeout&quot;] = 90
  
      # Obtain an OpenID Connect (OIDC) token from metadata server or using service account.
      open_id_connect_token = id_token.fetch_id_token(Request(), client_id)
      logging.info(f&quot;Retrieved open id connect (bearer) token {open_id_connect_token}&quot;)
  
      # Fetch the Identity-Aware Proxy-protected URL, including an authorization header containing &quot;Bearer &quot; followed by a
      # Google-issued OpenID Connect token for the service account.
      resp = requests.request(method, url, headers={&quot;Authorization&quot;: f&quot;Bearer {open_id_connect_token}&quot;}, **kwargs)
  
      if resp.status_code == 403:
          raise Exception(&quot;Service account does not have permission to access the IAP-protected application.&quot;)
      elif resp.status_code != 200:
          raise Exception(f&quot;Bad response from application: {resp.status_code} / {resp.headers} / {resp.text}&quot;)
      else:
          logging.info(f&quot;Response status - {resp.status_code}&quot;)
          return resp.json
  </code></pre>
  <p>This is the code that runs in the GC-functions
  I have checked the environment details in <strong>dlkpipelinesv1</strong> and <strong>piplines</strong> respectively, using this code :</p>
  <pre><code>credentials, _ = google.auth.default(
      scopes=['https://www.googleapis.com/auth/cloud-platform'])
  authed_session = google.auth.transport.requests.AuthorizedSession(
      credentials)
  
  # project_id = 'YOUR_PROJECT_ID'
  # location = 'us-central1'
  # composer_environment = 'YOUR_COMPOSER_ENVIRONMENT_NAME'
  
  environment_url = (
      'https://composer.googleapis.com/v1beta1/projects/{}/locations/{}'
      '/environments/{}').format(project_id, location, composer_environment)
  composer_response = authed_session.request('GET', environment_url)
  environment_data = composer_response.json()
  </code></pre>
  <p>and the two are using the same service accounts to run, i.e. the same IAM roles. Although I have noticed the following different details :</p>
  <p>In the old environment :</p>
  <pre><code>&quot;airflowUri&quot;: &quot;https://p5&lt;hidden_value&gt;-tp.appspot.com&quot;,
      &quot;privateEnvironmentConfig&quot;: { &quot;privateClusterConfig&quot;: {} },
  </code></pre>
  <p>in the new environment:</p>
  <pre><code>&quot;airflowUri&quot;: &quot;https://da&lt;hidden_value&gt;-tp.appspot.com&quot;,
      &quot;privateEnvironmentConfig&quot;: {
        &quot;privateClusterConfig&quot;: {},
        &quot;webServerIpv4CidrBlock&quot;: &quot;&lt;hidden_value&gt;&quot;,
        &quot;cloudSqlIpv4CidrBlock&quot;: &quot;&lt;hidden_value&gt;&quot;
      }
  </code></pre>
  <p>The service account that I use to make the post request has the following roles :</p>
  <pre><code>Cloud Functions Service Agent 
  Composer Administrator 
  Composer User
  Service Account Token Creator 
  Service Account User
  </code></pre>
  <p>The service account that runs my composer environment has the following roles :</p>
  <pre><code>BigQuery Admin
  Composer Worker
  Service Account Token Creator
  Storage Object Admin
  </code></pre>
  <p>But I am still receiving a <code>403 - Forbidden</code> in the <code>Log Explorer</code> when the <code>post</code> request is made to the airflow API.</p>
  <p>EDIT 2020-11-16 :</p>
  <p>I've updated to the latest <code>make_iap_request</code> code.
  I tinkered with the IAP within the security service, but I cannot find the webserver that will accept <code>HTTP: post</code> requests from my cloud functions... See the image bellow, anyway I added the service account to the default and CRM IAP resources just in case, but I still get this error :</p>
  <pre><code>Exception: Service account does not have permission to access the IAP-protected application.
  </code></pre>
  <p>The main question is: What IAP is at stake here?? And how do I add my service account as a user of this IAP.</p>
  <p>What am I missing?</p>
  <p><a href="https://i.stack.imgur.com/xiM4j.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/xiM4j.png" alt="List of HTTP IAP" /></a></p>