The previous version covered too few use cases, more specifically a single subscription. That is of course not the case for many hosts, so osbuild needs to understand subscriptions. When running org.osbuild.curl source, read the /etc/yum.repos.d/redhat.repo file and load the system subscriptions from there. While processing each url, guess which subscription is tied to the url and use the CA certificate, client certificate, and client key associated with this subscription. It must be done this way because the depsolving and fetching of RPMs may be performed on different hosts and the subscription credentials are different in such case. More detailed description of why this approach was chosen is available in osbuild-composer git: https://github.com/osbuild/osbuild-composer/pull/1405
67 lines
2.4 KiB
Python
67 lines
2.4 KiB
Python
"""Red Hat Subscription Manager support module
|
|
|
|
This module implements utilities that help with interactions
|
|
with the subscriptions attached to the host machine.
|
|
"""
|
|
|
|
import configparser
|
|
import re
|
|
|
|
|
|
class Subscriptions:
|
|
def __init__(self, repositories):
|
|
self.repositories = repositories
|
|
|
|
@classmethod
|
|
def from_host_system(cls):
|
|
"""Read redhat.repo file and process the list of repositories in there."""
|
|
with open("/etc/yum.repos.d/redhat.repo", "r") as fp:
|
|
return cls.parse_repo_file(fp)
|
|
|
|
@staticmethod
|
|
def _process_baseurl(input_url):
|
|
"""Create a regex from a baseurl.
|
|
|
|
The osbuild manifest format does not contain information about repositories.
|
|
It only includes URLs of each RPM. In order to make this RHSM support work,
|
|
osbuild needs to find a relation between a "baseurl" in a *.repo file and the
|
|
URL given in the manifest. To do so, it creates a regex from all baseurls
|
|
found in the *.repo file and matches them against the URL.
|
|
"""
|
|
# First escape meta characters that might occur in a URL
|
|
input_url = re.escape(input_url)
|
|
|
|
# Now replace variables with regexes (see man 5 yum.conf for the list)
|
|
for variable in ["\\$releasever", "\\$arch", "\\$basearch", "\\$uuid"]:
|
|
input_url = input_url.replace(variable, "[^/]*")
|
|
|
|
return re.compile(input_url)
|
|
|
|
@classmethod
|
|
def parse_repo_file(cls, fp):
|
|
"""Take a file object and reads its content assuming it is a .repo file."""
|
|
parser = configparser.ConfigParser()
|
|
parser.read_file(fp)
|
|
|
|
repositories = dict()
|
|
for section in parser.sections():
|
|
current = {
|
|
"matchurl": cls._process_baseurl(parser.get(section, "baseurl"))
|
|
}
|
|
for parameter in ["sslcacert", "sslclientkey", "sslclientcert"]:
|
|
current[parameter] = parser.get(section, parameter)
|
|
|
|
repositories[section] = current
|
|
|
|
return cls(repositories)
|
|
|
|
def get_secrets(self, url):
|
|
for parameters in self.repositories.values():
|
|
if parameters["matchurl"].match(url) is not None:
|
|
return {
|
|
"ssl_ca_cert": parameters["sslcacert"],
|
|
"ssl_client_key": parameters["sslclientkey"],
|
|
"ssl_client_cert": parameters["sslclientcert"]
|
|
}
|
|
|
|
raise RuntimeError(f"There are no RHSM secret associated with {url}")
|