# Copyright (c) 2025 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the COPYING file.
# Generated by the protocol buffer compiler. DO NOT EDIT!
# sources: arista/auditlog.v1/auditlog.proto, arista/auditlog.v1/services.gen.proto
# plugin: python-aristaproto
# This file has been @generated
__all__ = (
"Severity",
"CategoryType",
"Attributes",
"AuditlogKey",
"Auditlog",
"Filter",
"Category",
"MetaResponse",
"AuditlogRequest",
"AuditlogResponse",
"AuditlogSomeRequest",
"AuditlogSomeResponse",
"AuditlogStreamRequest",
"AuditlogStreamResponse",
"AuditlogServiceStub",
"AuditlogServiceBase",
)
from dataclasses import dataclass
from datetime import datetime
from typing import (
TYPE_CHECKING,
AsyncIterator,
Dict,
List,
Optional,
)
import aristaproto
import grpclib
from aristaproto.grpc.grpclib_server import ServiceBase
if TYPE_CHECKING:
import grpclib.server
from aristaproto.grpc.grpclib_client import MetadataLike
from grpclib.metadata import Deadline
[docs]
class Severity(aristaproto.Enum):
"""Severity is the level of criticality of the log."""
UNSPECIFIED = 0
"""SEVERITY_UNSPECIFIED severity"""
EMERGENCY = 1
"""SEVERITY_EMERGENCY severity"""
ALERT = 2
"""SEVERITY_ALERT severity"""
CRITICAL = 3
"""SEVERITY_CRITICAL severity"""
ERROR = 4
"""SEVERITY_ERROR severity"""
WARNING = 5
"""SEVERITY_WARNING severity"""
NOTICE = 6
"""SEVERITY_NOTICE severity"""
INFO = 7
"""SEVERITY_INFO severity"""
DEBUG = 8
"""SEVERITY_DEBUG severity"""
[docs]
class CategoryType(aristaproto.Enum):
"""CategoryType is the type of the category"""
UNSPECIFIED = 0
"""CATEGORY_TYPE_UNSPECIFIED categoryType"""
INVENTORY = 1
"""
CATEGORY_TYPE_INVENTORY categoryType
the identifier for this category is the deviceId
"""
CONFIGLET = 2
"""
CATEGORY_TYPE_CONFIGLET categoryType
the identifier for this category is a custom identifier defined by the user.
"""
IMAGE = 3
"""
CATEGORY_TYPE_IMAGE categoryType
the identifier for this category is a custom identifier defined by the user.
"""
LABEL = 4
"""
CATEGORY_TYPE_LABEL categoryType
the identifier for this category is a custom identifier defined by the user.
"""
USER = 5
"""
CATEGORY_TYPE_USER categoryType
the identifier for this category is a custom identifier defined by the user.
"""
CHANGECONTROL = 6
"""
CATEGORY_TYPE_CHANGECONTROL categoryType
the identifier for this category is a custom identifier defined by the user.
"""
ROLE = 7
"""
CATEGORY_TYPE_ROLE categoryType
the identifier for this category is a custom identifier defined by the user.
"""
SNAPSHOT = 8
"""
CATEGORY_TYPE_SNAPSHOT categoryType
the identifier for this category is a custom identifier defined by the user.
"""
EXECACTION = 9
"""
CATEGORY_TYPE_EXECACTION categoryType
the identifier for this category is a custom identifier defined by the user.
"""
ACCESSCONTROL = 10
"""
CATEGORY_TYPE_ACCESSCONTROL categoryType
the identifier for this category is a custom identifier defined by the user.
"""
[docs]
@dataclass(eq=False, repr=False)
class Attributes(aristaproto.Message):
"""Attributes are fields that make up the audit log."""
user: Optional[str] = aristaproto.message_field(1, wraps=aristaproto.TYPE_STRING)
"""user is the CloudVision user to which the log corresponds to."""
timestamp: datetime = aristaproto.message_field(2)
"""
timestamp at which the log was generated.
eg: 2006-01-02T15:04:05.000Z
"""
severity: "Severity" = aristaproto.enum_field(3)
"""
severity is level of importance or urgency of the audit log message.
eg: Debug, Info, Error
"""
device_name: Optional[str] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_STRING
)
"""device_name of the device for which audit log is sent."""
service: Optional[str] = aristaproto.message_field(5, wraps=aristaproto.TYPE_STRING)
"""
service name for which audit log is added.
eg: aaa, accesscontrol, inventory
"""
message: Optional[str] = aristaproto.message_field(6, wraps=aristaproto.TYPE_STRING)
"""
message in the audit log.
eg: \"user logged in\"
"""
tags: "___fmp__.MapStringString" = aristaproto.message_field(7)
"""
tags associated with the audit log.
eg: `tags: {mnemonic: DEVICE_PROVISIONING_SUCCESS}`
"""
[docs]
@dataclass(eq=False, repr=False)
class AuditlogKey(aristaproto.Message):
"""AuditlogKey is the unique key for each message."""
id: Optional[str] = aristaproto.message_field(1, wraps=aristaproto.TYPE_STRING)
"""id is the unique uuid for an audit log."""
[docs]
@dataclass(eq=False, repr=False)
class Auditlog(aristaproto.Message):
"""Auditlog contains the different fields as well as the formatted log."""
key: "AuditlogKey" = aristaproto.message_field(1)
"""key is the unique key for each message."""
attributes: "Attributes" = aristaproto.message_field(2)
"""attributes are the different fields that make up the audit log."""
log: Optional[str] = aristaproto.message_field(3, wraps=aristaproto.TYPE_STRING)
"""
log is the formatted log message containing the different attributes
including timestamp, serviceName, severity , userName in addition to
the message.
May 17 12:50:54 host service Message: user logged in, User: abc
"""
[docs]
@dataclass(eq=False, repr=False)
class Filter(aristaproto.Message):
"""Filter are the different filters that can be applied."""
category: "Category" = aristaproto.message_field(1)
"""
category filters based on category of the logs.
eg:
`{filter:[{category:{custom_type: abc}}]}`
filters all logs associated with the service abc.
"""
queries: "___fmp__.RepeatedString" = aristaproto.message_field(2)
"""
queries is the list of strings that should be present in the log.
eg: `{filter:[{queries:{values:[abc]}}]}` searches for all
logs containing the string abc.
"""
tags: "___fmp__.MapStringString" = aristaproto.message_field(3)
"""
tags filters based on tags associated with the log.
eg: `{filter:[{tags:{values:{id:xyz}}}]}`
searches for all logs associates with the tag
id:xyz.
"""
severity: "Severity" = aristaproto.enum_field(4)
"""severity is level of importance or urgency of the audit log message."""
user: Optional[str] = aristaproto.message_field(5, wraps=aristaproto.TYPE_STRING)
"""user filters the audit logs by given username across categories."""
[docs]
@dataclass(eq=False, repr=False)
class Category(aristaproto.Message):
"""
Category organizes audit logs into different groups using the CategoryType and the identifier
eg: `filter:[{category:{type:CATEGORY_TYPE_INVENTORY,identifier:abc}}]`
filters all logs associated with the service \"inventory\" and device abc
"""
type: "CategoryType" = aristaproto.enum_field(1)
"""type of the category."""
identifier: Optional[str] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_STRING
)
"""
identifier is the object with which we want to filter the category.
eg: fetch all logs from the device with id abc
Here the category is inventory and the identifier is abc.
"""
custom_type: Optional[str] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_STRING
)
"""
custom_type is a custom category that can be created by the user.
this is used if none of the existing categories can support the users's usecase.
"""
[docs]
@dataclass(eq=False, repr=False)
class AuditlogRequest(aristaproto.Message):
""" """
key: "AuditlogKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a Auditlog instance to retrieve.
This value must be populated.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class AuditlogResponse(aristaproto.Message):
""" """
value: "Auditlog" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time carries the (UTC) timestamp of the last-modification of the
Auditlog instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class AuditlogSomeRequest(aristaproto.Message):
""" """
keys: List["AuditlogKey"] = aristaproto.message_field(1)
"""
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class AuditlogSomeResponse(aristaproto.Message):
""" """
value: "Auditlog" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
Auditlog instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class AuditlogStreamRequest(aristaproto.Message):
""" """
filter: List["Filter"] = aristaproto.message_field(2)
"""
For each Auditlog in the list, all populated fields are considered ANDed together
as a filtering operation. Similarly, the list itself is ORed such that any individual
filter that matches a given Auditlog is streamed to the user.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each Auditlog at end.
* Each Auditlog response is fully-specified (all fields set).
* start: Returns the state of each Auditlog at start, followed by updates until now.
* Each Auditlog response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Auditlog at start, followed by updates
until end.
* Each Auditlog response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class AuditlogStreamResponse(aristaproto.Message):
""" """
value: "Auditlog" = aristaproto.message_field(1)
"""
Value is a value deemed relevant to the initiating request.
This structure will always have its key-field populated. Which other fields are
populated, and why, depends on the value of Operation and what triggered this notification.
"""
time: datetime = aristaproto.message_field(2)
"""Time holds the timestamp of this Auditlog's last modification."""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the Auditlog value in this response should be considered.
Under non-subscribe requests, this value should always be INITIAL. In a subscription,
once all initial data is streamed and the client begins to receive modification updates,
you should not see INITIAL again.
"""
[docs]
class AuditlogServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
auditlog_request: "AuditlogRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AuditlogResponse":
""" """
return await self._unary_unary(
"/arista.auditlog.v1.AuditlogService/GetOne",
auditlog_request,
AuditlogResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
auditlog_some_request: "AuditlogSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[AuditlogSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.auditlog.v1.AuditlogService/GetSome",
auditlog_some_request,
AuditlogSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
auditlog_stream_request: "AuditlogStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[AuditlogStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.auditlog.v1.AuditlogService/GetAll",
auditlog_stream_request,
AuditlogStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
auditlog_stream_request: "AuditlogStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[AuditlogStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.auditlog.v1.AuditlogService/Subscribe",
auditlog_stream_request,
AuditlogStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
from .... import fmp as ___fmp__
from ... import subscriptions as __subscriptions__
from ... import time as __time__
[docs]
class AuditlogServiceBase(ServiceBase):
""" """
[docs]
async def get_one(self, auditlog_request: "AuditlogRequest") -> "AuditlogResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, auditlog_some_request: "AuditlogSomeRequest"
) -> AsyncIterator[AuditlogSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, auditlog_stream_request: "AuditlogStreamRequest"
) -> AsyncIterator[AuditlogStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, auditlog_stream_request: "AuditlogStreamRequest"
) -> AsyncIterator[AuditlogStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self, stream: "grpclib.server.Stream[AuditlogRequest, AuditlogResponse]"
) -> None:
request = await stream.recv_message()
response = await self.get_one(request)
await stream.send_message(response)
async def __rpc_get_some(
self, stream: "grpclib.server.Stream[AuditlogSomeRequest, AuditlogSomeResponse]"
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_some,
stream,
request,
)
async def __rpc_get_all(
self,
stream: "grpclib.server.Stream[AuditlogStreamRequest, AuditlogStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all,
stream,
request,
)
async def __rpc_subscribe(
self,
stream: "grpclib.server.Stream[AuditlogStreamRequest, AuditlogStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe,
stream,
request,
)
async def __rpc_get_meta(
self, stream: "grpclib.server.Stream[AuditlogStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
response = await self.get_meta(request)
await stream.send_message(response)
async def __rpc_subscribe_meta(
self, stream: "grpclib.server.Stream[AuditlogStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.auditlog.v1.AuditlogService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
AuditlogRequest,
AuditlogResponse,
),
"/arista.auditlog.v1.AuditlogService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
AuditlogSomeRequest,
AuditlogSomeResponse,
),
"/arista.auditlog.v1.AuditlogService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
AuditlogStreamRequest,
AuditlogStreamResponse,
),
"/arista.auditlog.v1.AuditlogService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
AuditlogStreamRequest,
AuditlogStreamResponse,
),
"/arista.auditlog.v1.AuditlogService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
AuditlogStreamRequest,
MetaResponse,
),
"/arista.auditlog.v1.AuditlogService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
AuditlogStreamRequest,
MetaResponse,
),
}