Pomerium

# Programmatic access

This page describes how to obtain Pomerium access credentials programmatically via a web-based oauth2 style authorization flow. If you have ever used Google's gcloud commandline app, the mechanism is very similar.

# Components

# Login API

The API returns a cryptographically signed sign-in url that can be used to complete a user-driven login process with Pomerium and your identity provider. The login API endpoint takes a redirect_uri query parameter as an argument which points to the location of the callback server to be called following a successful login.

For example:

$ curl "https://verify.example.com/.pomerium/api/v1/login?redirect_uri=http://localhost:8000"

https://authenticate.example.com/.pomerium/sign_in?redirect_uri=http%3A%2F%2Flocalhost%3Fpomerium_callback_uri%3Dhttps%253A%252F%verify.corp.example%252F.pomerium%252Fapi%252Fv1%252Flogin%253Fredirect_uri%253Dhttp%253A%252F%252Flocalhost&sig=hsLuzJctmgsN4kbMeQL16fe_FahjDBEcX0_kPYfg8bs%3D&ts=1573262981

# Callback handler

It is the script or application's responsibility to create a HTTP callback handler. Authenticated sessions are returned in the form of a callback (opens new window) from pomerium to a HTTP server. This is the redirect_uri value used to build login API's URL, and represents the URL of a (usually local) HTTP server responsible for receiving the resulting user session in the form of pomerium_jwt query parameters.

See the python script below for example of how to start a callback server, and store the session payload.

# Handling expiration and revocation

Your script or application should anticipate the possibility that your underlying refresh_token may stop working. For example, a refresh token might stop working if the underlying user changes passwords, revokes access, or if the administrator removes rotates or deletes the OAuth Client ID.

# High level workflow

The application interacting with Pomerium must manage the following workflow. Consider the following example where a script or program desires delegated, programmatic access to the domain verify.corp.domain.example:

  1. The script or application requests a new login url from the pomerium managed endpoint (e.g. https://verify.corp.domain.example/.pomerium/api/v1/login) and takes a redirect_uri as an argument.
  2. The script or application opens a browser or redirects the user to the returned login page.
  3. The user completes the identity providers login flow.
  4. The identity provider makes a callback to pomerium's authenticate service (e.g. authenticate.corp.domain.example) .
  5. Pomerium's authenticate service creates a user session and redirect token, then redirects back to the managed endpoint (e.g. verify.corp.domain.example)
  6. Pomerium's proxy service makes a callback request to the original redirect_uri with the user session and as an argument.
  7. The script or application is responsible for handling that http callback request, and securely handling the callback session (pomerium_jwt) queryparam.
  8. The script or application can now make any requests as normal to the upstream application by setting the Authorization: Pomerium ${pomerium_jwt} header.

# Example Code

Please consider see the following minimal but complete python example.

python3 scripts/programmatic_access.py \
	--dst https://verify.example.com/headers

from __future__ import absolute_import, division, print_function

import argparse
import http.server
import json
import sys
import urllib.parse
import webbrowser
from urllib.parse import urlparse
import requests

done = False

parser = argparse.ArgumentParser()
parser.add_argument("--login", action="store_true")
parser.add_argument(
    "--dst", default="https://verify.example.com/json",
)
parser.add_argument("--server", default="localhost", type=str)
parser.add_argument("--port", default=8000, type=int)
parser.add_argument(
    "--cred", default="pomerium-cred.json",
)
args = parser.parse_args()


class PomeriumSession:
    def __init__(self, jwt):
        self.jwt = jwt

    def to_json(self):
        return json.dumps(self.__dict__, indent=2)

    @classmethod
    def from_json_file(cls, fn):
        with open(fn) as f:
            data = json.load(f)
            return cls(**data)


class Callback(http.server.BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        # silence http server logs for now
        return

    def do_GET(self):
        global args
        global done
        self.send_response(200)
        self.end_headers()
        response = b"OK"
        if "pomerium" in self.path:
            path = urllib.parse.urlparse(self.path).query
            path_qp = urllib.parse.parse_qs(path)
            session = PomeriumSession(
                path_qp.get("pomerium_jwt")[0],
            )
            done = True
            response = session.to_json().encode()
            with open(args.cred, "w", encoding="utf-8") as f:
                f.write(session.to_json())
                print("=> pomerium json credential saved to:\n{}".format(f.name))

        self.wfile.write(response)


def main():
    global args

    dst = urllib.parse.urlparse(args.dst)
    try:
        cred = PomeriumSession.from_json_file(args.cred)
    except:
        print("=> no credential found, let's login")
        args.login = True

    # initial login to make sure we have our credential
    if args.login:
        dst = urllib.parse.urlparse(args.dst)
        query_params = {
            "pomerium_redirect_uri": "http://{}:{}".format(args.server, args.port)
        }
        enc_query_params = urllib.parse.urlencode(query_params)
        dst_login = "{}://{}{}?{}".format(
            dst.scheme, dst.hostname, "/.pomerium/api/v1/login", enc_query_params,
        )
        response = requests.get(dst_login)
        print("=> Your browser has been opened to visit:\n{}".format(response.text))
        webbrowser.open(response.text)

        with http.server.HTTPServer((args.server, args.port), Callback) as httpd:
            while not done:
                httpd.handle_request()

    cred = PomeriumSession.from_json_file(args.cred)
    response = requests.get(
        args.dst,
        headers={
            "Authorization": "Pomerium {}".format(cred.jwt),
            "Content-type": "application/json",
            "Accept": "application/json",
        },
    )
    print(
        "==> request\n{}\n==> response.status_code\n{}\n==>response.text\n{}\n".format(
            args.dst, response.status_code, response.text
        )
    )


if __name__ == "__main__":
    main()

← TCP Support →