Viewing file:  imunify_patch_subscription.py (2.34 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
Copyright © 2019 Cloud Linux Software Inc.
This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import json
import logging
from urllib.parse import urljoin
from urllib.request import Request
from imav.malwarelib.model import ImunifyPatchSubscription
from defence360agent.api.server import API, APIError
from defence360agent.internals.iaid import (
IndependentAgentIDAPI,
IAIDTokenError,
)
logger = logging.getLogger(__name__)
class ImunifyPatchSubscriptionAPI(API):
URL = urljoin(API._BASE_URL, "/api/patch/subscriptions/check")
@classmethod
async def get_subscriptions(cls, ids: list[str]) -> list[str] | None:
"""Get active subscriptions for specific ids.
Return None if interaction with the remote API failed.
"""
if not ids:
return []
try:
token = await IndependentAgentIDAPI.get_token()
except IAIDTokenError as exc:
logger.error(
"Can't get iaid token: %s. Return empty subscription list.",
exc,
)
return None
request = Request(
cls.URL,
data=json.dumps({"users": ids}).encode(),
headers={"X-Auth": token, "Content-Type": "application/json"},
)
try:
result = await cls.async_request(request)
except APIError as exc:
logger.error(
"Failed to get subscriptions details: %s. "
"Return empty subscription list.",
exc,
)
return None
return result["patchActive"]
def has_imunify_patch_subscriptions(_: str) -> bool:
return ImunifyPatchSubscription.select().exists()
|