How to set the data HTTP request headers

[pepjo]

Hi everyone,

I've been trying to figure out how to set a specific HTTP header when TiTiler queries a STAC API. In my use case, I would like to set it to whatever the header is when TiTiler is called.

I have been able to read the Authorization header that is used when calling TiTiler by using:

# Add private COG endpoints requiring token validation
from fastapi import Security, Query
from fastapi.security.api_key import APIKeyHeader

from titiler.application.main import app
from titiler.core.factory import TilerFactory

api_token_header = APIKeyHeader(name="Authorization", auto_error=False)

# Here I can access the token and modify the path, but not the headers :(
def DatasetPathParams(
    url: str = Query(..., description="Dataset URL"),
    access_token: str = Security(api_token_header),
) -> str:
    """Create dataset path from args"""
    print("URL")
    print(url)
    print("TOKEN")
    print(access_token)
    return url

# Custom router with token dependency
tiler = TilerFactory(path_dependency=DatasetPathParams)

app.include_router(tiler.router, prefix="/test", tags=["Test"])

In this DatasetPathParams function, I would be able to modify the URL used to get the STAC item. However, I am not interested in changing the URL but the headers that will be used.

Is there a way to achieve what I would like to do?

Thanks in advance for your help

[vincentsarago]

👋 @pepjo

To modify the headers use for the STAC request you'll need to set fetch_options={"headers": {YOUR HEADERS VALUE}} rio-tiler

Sadly this has to be pass at reader init

titiler/src/titiler/core/titiler/core/factory.py

Line 212 in ac184ae

with self.reader(src_path, **self.reader_options) as src_dst:

but we don't have a dynamic way of populating the reader options 🙁. I fear you'll have to re-write the routes where you want to use this.

@self.router.get(
    "/info",
    response_model=Info,
    response_model_exclude_none=True,
    response_class=JSONResponse,
    responses={200: {"description": "Return dataset's basic info."}},
)
def info(
    src_path=Depends(self.path_dependency),
    access_token: str = Security(api_token_header),
    kwargs: Dict = Depends(self.additional_dependency),
):
    """Return dataset's basic info."""
    with self.reader(
        src_path,
        fetch_options={
            "headers": {
                "Authorization": access_token,
            }
        },
        **self.reader_options,
    ) as src_dst:
        return src_dst.info(**kwargs)

An other solution would maybe to get rid of the static reader_options and replace it with a Dependency injection for each endpoint, but that will require some work 🤷‍♂️

[pepjo]

Hi @vincentsarago !

Nice to e-meet you and thank you very much for the quick and useful reply.

I will have a chat with the team to dicuss how we implement this. In this case, it might be easier for us to pass the token as a query parameter and modify our STAC server to read the token from there.

However, if we decide to modify the endpoints to include the option as you mention. Is this something you would be interested in merging to TiTiler? If you are and we do it, we will make sure to create a pull request ☺️.

[vincentsarago]

You could also create a custom STAC reader like that decode an input URL

import attr
from rio_tiler.io import STACReader


@attr.s
class MyCustomSTACReader(STACReader):
    """Custom STACReader."""

    def __attrs_post_init__(self):
        """Decode data: urls."""
        # for the example I'll consider that you pass "{stac item url}|{token}"
        path, token = self.input.split("|") # if you are using older titiler (not 0.4) this should be `self.filepath`
        self.input = path. # if you are using older titiler (not 0.4) this should be `self.filepath`
        self.fetch_options = {"headers": {"Authorization": token}}

        super().__attrs_post_init__()


def DatasetPathParams(
    url: str = Query(..., description="Dataset URL"),
    access_token: str = Security(api_token_header),
) -> str:
    """Create dataset path from args"""
    return f"{url}|{access_token}"


# Custom router with token dependency
tiler = TilerFactory(path_dependency=DatasetPathParams, reader=MyCustomSTACReader)

[vincentsarago]

Well there is also another less hacky solution

from typing import Tuple
import attr
from rio_tiler.io import STACReader
from titiler.core.factory import MultiBaseTilerFactory


@attr.s
class MyCustomSTACReader(STACReader):
    """Custom STACReader."""

    def __attrs_post_init__(self):
        """Decode data: urls."""
        path, token = self.input # if you are using older titiler (not 0.4) this should be `self.filepath`
        self.input = path. # if you are using older titiler (not 0.4) this should be `self.filepath`
        self.fetch_options = {"headers": {"Authorization": token}}

        super().__attrs_post_init__()


def DatasetPathParams(
    url: str = Query(..., description="Dataset URL"),
    access_token: str = Security(api_token_header),
) -> Tuple[str, str]:
    """Create dataset path from args"""
    return (url, token)


# Custom router with token dependency
tiler = MultiBaseTilerFactory(path_dependency=DatasetPathParams, reader=MyCustomSTACReader) 

Mypy might not be happy with some type information but it should work in theory

[guillemc23]

How would you do this in case I just need to propagate the request headers? For example, my frontend sends some authorization headers and i want to use these to read the STAC item as well from the catalog. I haven't found a way with this response but I think it should be somewhat similar. Thank you!

This works but I couldn't find a way to set these headers dynamically from the ones that come with the request

headers = {
    "Authorization": "Bearer xxx"
}


@attr.s
class CustomSTACReader(STACReader):
    """Custom STACReader."""

    def __attrs_post_init__(self):
        self.fetch_options = {"headers": headers}

        super().__attrs_post_init__()

[vincentsarago]

You should be able to create a reader_dependency https://github.com/developmentseed/titiler/blob/main/src/titiler/core/titiler/core/factory.py#L262

[vincentsarago]

I've proposed 2 different solutions, I'll mark this as answered. @pepjo feel free to comment if it's not what you expected 🙏

[pepjo]

Sorry about my late reply, I was off sick and I missed the messages :(

Those suggestions were really good! Thank you for those, they really solve the problem we had, the second in particular looks especially clean and simple.

Thanks for your help on this @vincentsarago 😊

Best,

[vincentsarago]

Hope you are feeling better now 🙏

just found a typo in the second example, it should be MultiBaseTilerFactory and not TilerFactory

tiler = MultiBaseTilerFactory(path_dependency=DatasetPathParams, reader=MyCustomSTACReader)

[pepjo]

Makes sense, since this only applies to the STAC endpoints, thank you!

[guillemc23]

I'll leave here my approach to forward authorization headers from the frontend to a STAC API, just in case anybody needs to handle that use case:

from typing import Any, Dict, Optional

import attr
from fastapi import Request
from titiler.core.dependencies import DefaultDependency

@attr.dataclass
class HeaderDependency(DefaultDependency):
    fetch_options: Optional[Dict[str, Any]] = None

    @classmethod
    async def from_request(cls, request: Request):
        """Instantiate object from a FastAPI Request."""
        headers = {
            "headers": {
                "authorization": request.headers.get("authorization", ""),
            }
        }
        return cls(fetch_options=headers)


async def get_header_dependency(request: Request) -> HeaderDependency:
    """FastAPI Dependency that creates a `HeaderDependency` object from an incoming request."""
    return await HeaderDependency.from_request(request)

######
# Add STAC extension with custom dependency
stac = MultiBaseTilerFactory(
    reader=STACReader,
    router_prefix="/stac",
    reader_dependency=get_header_dependency,  # Passes incoming headers to STAC API
)