Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[1.28] CCT-69: Anonymous registration #3383

Draft
wants to merge 19 commits into
base: subscription-manager-1.28
Choose a base branch
from
Draft
Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Implement API endpoints for Automatic registration v2
* Card ID: CCT-67 (main branch)
* Card ID: CCT-69 (backport)

- /cloud/authorize?version=2: New
- /consumers/{uuid}/certificates: Added 'Authorization' header

(Cherry-picked from 43f57ca)
  • Loading branch information
m-horky authored and ptoscano committed Oct 2, 2024
commit 3616a69c9dabdf58d94a86cc233f29391f2a61c4
45 changes: 41 additions & 4 deletions src/rhsm/connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
import sys
import time
import traceback
from typing import Optional
from typing import Any, Dict, List, Optional
from pathlib import Path

from email.utils import format_datetime
Expand Down Expand Up @@ -1180,6 +1180,30 @@ def getJWToken(self, cloud_id, metadata, signature):
headers=headers
)

def getCloudJWT(self, cloud_id: str, metadata: str, signature: str) -> Dict[str, Any]:
"""Obtain cloud JWT.

This method is part of the Cloud registration v2: standard or anonymous flow.

:param cloud_id: Cloud provider, e.g. 'aws', 'azure' or 'gcp'.
:param metadata: Base64 encoded public cloud metadata.
:param signature: Base64 encoded public cloud signature.
"""
data = {
"type": cloud_id,
"metadata": metadata,
"signature": signature,
}
headers = {
"Content-Type": "application/json",
}

return self.conn.request_post(
method="/cloud/authorize?version=2",
params=data,
headers=headers,
)

def registerConsumer(self, name="unknown", type="system", facts={},
owner=None, environments=None, keys=None,
installed_products=None, uuid=None, hypervisor_id=None,
Expand Down Expand Up @@ -1514,16 +1538,29 @@ def unregisterConsumer(self, consumerId):
method = '/consumers/%s' % self.sanitize(consumerId)
return self.conn.request_delete(method)

def getCertificates(self, consumer_uuid, serials=[]):
def getCertificates(
self,
consumer_uuid: str,
serials: Optional[list] = None,
jwt: Optional[str] = None,
) -> List[dict]:
"""
Fetch all entitlement certificates for this consumer.
Specify a list of serial numbers to filter if desired.

:param consumer_uuid: consumer UUID
:param serials: list of entitlement serial numbers
:param jwt: JWT identifying an anonymous system
"""
method = '/consumers/%s/certificates' % (self.sanitize(consumer_uuid))
if len(serials) > 0:
if serials:
serials_str = ','.join(serials)
method = "%s?serials=%s" % (method, serials_str)
return self.conn.request_get(method)
headers = {}
if jwt:
headers["Authorization"] = "Bearer {jwt}".format(jwt=jwt)

return self.conn.request_get(method, headers=headers)

def getCertificateSerials(self, consumerId):
"""
Expand Down