admin/its-a-feature-Mythic
1
0
Fork 0
mirror of https://github.com/its-a-feature/Mythic synced 2026-06-08 14:55:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

updating event log message handling for performance updates

Browse Source
This commit is contained in:
its-a-feature
2026-04-13 09:08:13 -05:00
parent fa235b6378
commit e1806f0fd5
15 changed files with 246 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.MD
+9
View File
@@ -4,6 +4,15 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [3.4.34] - 2026-04-13
### Changed
- Updated way that SendAllOperationsMessage functions to reduce duplicate messages to event log
- Reduced the number of lookups per error message to increment count by caching error lookup data
- Updated event log messages to ensure proper UTF8 encoding (and removal of bad characters)
- Using channels to ensure no duplicate messages are created due to go concurrency
## [3.4.33] - 2026-04-10
### Changed
VERSION
+1 -1
View File
@@ -1 +1 @@
3.4.33
3.4.34
mythic-docker/src/database/utils.go
-29
View File
@@ -165,35 +165,6 @@ func HashUserPassword(databaseOperator databaseStructs.Operator, password string
return fmt.Sprintf("%x", hashBytes)
}
func ResolveAllOperationsMessage(message string, operationID int) {
/*
Resolve a message in all operation's event logs if operationID is 0, otherwise just resolve it to the specific operation.
*/
operations := []databaseStructs.Operation{}
if err := DB.Select(&operations, `SELECT id FROM operation WHERE complete=false`); err != nil {
logging.LogError(err, "Failed to get operations for ResolveAllOperationsMessage", "data", message)
return
}
for _, operation := range operations {
if operationID == 0 || operation.ID == operationID {
// this is the operation we're interested in
updateObject := databaseStructs.Operationeventlog{
Message: message,
OperationID: operationID,
Level: MESSAGE_LEVEL_INFO,
}
if operationID == 0 {
updateObject.OperationID = operation.ID
}
if _, err := DB.NamedExec(`UPDATE operationeventlog SET
resolved=true
WHERE warning=true AND resolved=false AND deleted=false AND message=:message AND operation_id=:operation_id`, updateObject); err != nil {
logging.LogError(err, "Failed to resolve message")
}
}
}
}
func AssignNewOperatorAllBrowserScripts(userID int) {
browserscripts := []databaseStructs.Browserscript{}
if err := DB.Select(&browserscripts, `SELECT
mythic-docker/src/rabbitmq/check_container_status.go
+5 -5
View File
@@ -300,7 +300,7 @@ func checkContainerStatus() {
0, fmt.Sprintf("%s_container_down", container), database.MESSAGE_LEVEL_INFO, true)
go updateDownContainerBuildingPayloads(container)
} else {
go database.ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go CreateGraphQLSpectatorAPITokenAndSendOnStartMessage(container)
}
} else {
@@ -332,7 +332,7 @@ func checkContainerStatus() {
getDownContainerMessage(container),
0, fmt.Sprintf("%s_container_down", container), database.MESSAGE_LEVEL_INFO, true)
} else {
go database.ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go CreateGraphQLSpectatorAPITokenAndSendOnStartMessage(container)
}
} else {
@@ -363,7 +363,7 @@ func checkContainerStatus() {
getDownContainerMessage(container),
0, fmt.Sprintf("%s_container_down", container), database.MESSAGE_LEVEL_INFO, true)
} else {
go database.ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go CreateGraphQLSpectatorAPITokenAndSendOnStartMessage(container)
}
} else {
@@ -395,7 +395,7 @@ func checkContainerStatus() {
getDownContainerMessage(container),
0, fmt.Sprintf("%s_container_down", container), database.MESSAGE_LEVEL_INFO, true)
} else {
go database.ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go CreateGraphQLSpectatorAPITokenAndSendOnStartMessage(container)
}
} else {
@@ -426,7 +426,7 @@ func checkContainerStatus() {
getDownContainerMessage(container),
0, fmt.Sprintf("%s_container_down", container), database.MESSAGE_LEVEL_INFO, true)
} else {
go database.ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(container), 0)
go CreateGraphQLSpectatorAPITokenAndSendOnStartMessage(container)
}
} else {
mythic-docker/src/rabbitmq/initialize.go
+6 -5
View File
@@ -86,8 +86,9 @@ var pushC2StreamingDisconnectNotification = make(chan int, 100)
func Initialize() {
RabbitMQConnection.channelMutexMap = make(map[string]*channelMutex)
go invalidateAllSpectatorAPITokens()
go InvalidateOperationEventLogCacheMap()
invalidateAllNonUserAPITokens()
InvalidateOperationEventLogCacheMap()
go listenForOperationsMessages()
go listenForWriteDownloadChunkToLocalDisk()
go listenForFileBrowserData()
go listenForAsyncAgentMessagePostResponseContent()
@@ -109,10 +110,10 @@ func Initialize() {
go listenForEvents()
go initializeEventGroupCronSchedulesOnStart()
// start listening for new messages from push c2 profiles, needs gRPC initialized first
go listenForPushConnectDisconnectMessages()
go processAgentMessageFromPushC2()
go interceptProxyDataToAgentForPushC2()
go checkIfActiveCallbacksAreAliveForever()
go listenForPushConnectDisconnectMessages()
go func() {
// wait 20s for things to stabilize a bit, then send a startup message
time.Sleep(time.Second * 30)
@@ -171,9 +172,9 @@ func emitStartupMessages() {
}
func invalidateAllSpectatorAPITokens() {
func invalidateAllNonUserAPITokens() {
_, err := database.DB.Exec(`UPDATE apitokens
SET deleted=true, active=true WHERE token_type=$1 OR token_type=$2 OR token_type=$3`,
SET deleted=true, active=false WHERE token_type=$1 OR token_type=$2 OR token_type=$3`,
mythicjwt.AUTH_METHOD_GRAPHQL_SPECTATOR, mythicjwt.AUTH_METHOD_TASK, mythicjwt.AUTH_METHOD_EVENT)
if err != nil {
logging.LogError(err, "failed to mark all spectator tokens as deleted and inactive")
mythic-docker/src/rabbitmq/recv_c2_sync.go
+1 -1
View File
@@ -191,7 +191,7 @@ func c2Sync(in C2SyncMessage) error {
return err
}
go SendAllOperationsMessage(fmt.Sprintf("Successfully synced %s with container version %s", c2Profile.Name, in.ContainerVersion), 0, "debug", database.MESSAGE_LEVEL_DEBUG, false)
go database.ResolveAllOperationsMessage(getDownContainerMessage(c2Profile.Name), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(c2Profile.Name), 0)
go autoStartC2Profile(c2Profile, false)
if newProfile {
go reSyncPayloadTypes()
mythic-docker/src/rabbitmq/recv_consuming_container_sync.go
+1 -1
View File
@@ -134,7 +134,7 @@ func consumingServicesSync(in ConsumingContainerSyncMessage) error {
}
go SendAllOperationsMessage(fmt.Sprintf("Successfully synced %s with container version %s",
consumingContainer.Name, in.ContainerVersion), 0, "debug", database.MESSAGE_LEVEL_DEBUG, false)
go database.ResolveAllOperationsMessage(getDownContainerMessage(consumingContainer.Name), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(consumingContainer.Name), 0)
checkContainerStatusAddConsumingContainerChannel <- consumingContainer
// update eventgroup consumingcontainer mappings
eventing.UpdateEventGroupConsumingContainersMappingByConsumingContainer(consumingContainer)
mythic-docker/src/rabbitmq/recv_custombrowser_sync.go
+1 -1
View File
@@ -190,7 +190,7 @@ func customBrowserSync(in CustomBrowserSyncMessage) error {
return err
}
go SendAllOperationsMessage(fmt.Sprintf("Successfully synced %s with container version %s", customSyncBrowser.Name, in.ContainerVersion), 0, "debug", database.MESSAGE_LEVEL_DEBUG, false)
go database.ResolveAllOperationsMessage(getDownContainerMessage(customSyncBrowser.Name), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(customSyncBrowser.Name), 0)
checkContainerStatusAddCustomBrowserChannel <- customSyncBrowser
go CreateGraphQLSpectatorAPITokenAndSendOnStartMessage(customSyncBrowser.Name)
mythic-docker/src/rabbitmq/recv_pt_sync.go
+1 -1
View File
@@ -450,7 +450,7 @@ func payloadTypeSync(in PayloadTypeSyncMessage) error {
InvalidateAllCachedUUIDInfo()
}
go SendAllOperationsMessage(fmt.Sprintf("Successfully synced %s with container version %s", payloadtype.Name, in.ContainerVersion), 0, "debug", database.MESSAGE_LEVEL_DEBUG, false)
go database.ResolveAllOperationsMessage(getDownContainerMessage(payloadtype.Name), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(payloadtype.Name), 0)
checkContainerStatusAddPtChannel <- payloadtype
if !in.ForcedSync {
go CreateGraphQLSpectatorAPITokenAndSendOnStartMessage(payloadtype.Name)
mythic-docker/src/rabbitmq/recv_tr_sync.go
+1 -1
View File
@@ -56,7 +56,7 @@ func processTrSyncMessages(msg amqp.Delivery) interface{} {
// successfully synced
response.Success = true
go SendAllOperationsMessage(fmt.Sprintf("Successfully synced %s with container version %s", trSyncMsg.Name, trSyncMsg.ContainerVersion), 0, "debug", database.MESSAGE_LEVEL_DEBUG, false)
go database.ResolveAllOperationsMessage(getDownContainerMessage(trSyncMsg.Name), 0)
go ResolveAllOperationsMessage(getDownContainerMessage(trSyncMsg.Name), 0)
logging.LogDebug("Successfully synced", "service", trSyncMsg.Name)
}
return response
mythic-docker/src/rabbitmq/utils.go
+216 -111
View File
@@ -1,6 +1,7 @@
package rabbitmq
import (
"bytes"
"database/sql"
"encoding/json"
"errors"
@@ -1245,6 +1246,8 @@ func GetSaveFilePath() (string, string, error) {
var operationsMap = make(map[int]databaseStructs.Operation)
var operationsMapMutex sync.RWMutex
var unresolvedErrors = make(map[int]map[string]map[string]int)
var lastUpdateTime = time.Now()
func InvalidateOperationEventLogCacheMap() {
operationsMapMutex.Lock()
@@ -1260,123 +1263,225 @@ func InvalidateOperationEventLogCacheMap() {
operationsMap[operation.ID] = operation
}
}
func SendAllOperationsMessage(message string, operationID int, source string, messageLevel database.MESSAGE_TYPE, warning bool) {
var operations []databaseStructs.Operation
operationsMapMutex.RLock()
if operationID == 0 {
for _, operation := range operationsMap {
operations = append(operations, operation)
}
} else {
operation, ok := operationsMap[operationID]
if !ok {
operationsMapMutex.RUnlock()
logging.LogError(nil, "unknown operation id, can't send message", "operation_id", operationID)
return
}
operations = append(operations, operation)
func checkUnresolvedError(operationID int, source string, level string) int {
if _, ok := unresolvedErrors[operationID]; !ok {
return 0
}
operationsMapMutex.RUnlock()
sourceString := source
if sourceString == "" {
sourceString = uuid.NewString()
if _, ok := unresolvedErrors[operationID][source]; !ok {
return 0
}
for _, operation := range operations {
if operationID == 0 || operation.ID == operationID {
// this is the operation we're interested in
if messageLevel == "warning" || warning {
existingMessage := databaseStructs.Operationeventlog{}
err := database.DB.Get(&existingMessage, `
SELECT id, count, "message", source, "level" FROM operationeventlog WHERE
warning=true and source=$1 and operation_id=$2 and resolved=false and deleted=false and "level"=$3
`, sourceString, operation.ID, messageLevel)
if !errors.Is(err, sql.ErrNoRows) && err != nil {
logging.LogError(err, "Failed to query existing event log message")
continue
}
if errors.Is(err, sql.ErrNoRows) {
if messageLevel == "warning" {
messageLevel = database.MESSAGE_LEVEL_INFO
}
newMessage := databaseStructs.Operationeventlog{
Source: sourceString,
Level: messageLevel,
Warning: warning,
Message: message,
OperationID: operation.ID,
Count: 1,
}
_, err = database.DB.NamedExec(`INSERT INTO operationeventlog
(source, "level", "message", operation_id, count, warning)
VALUES
(:source, :level, :message, :operation_id, :count, :warning)`, newMessage)
if err != nil {
logging.LogError(err, "Failed to create new operationeventlog message")
continue
}
go RabbitMQConnection.EmitWebhookMessage(WebhookMessage{
OperationID: operation.ID,
OperationName: operation.Name,
OperationWebhook: operation.Webhook,
OperationChannel: operation.Channel,
OperatorUsername: "",
Action: WEBHOOK_TYPE_ALERT,
Data: map[string]interface{}{
"message": newMessage.Message,
"source": newMessage.Source,
"count": newMessage.Count,
"timestamp": time.Now().UTC(),
},
})
EventingChannel <- EventNotification{
Trigger: eventing.TriggerAlert,
OperationID: operation.ID,
Outputs: map[string]interface{}{
"alert": newMessage.Message,
},
}
continue
}
// err was nil, so we did get a matching existing message
existingMessage.Count += 1
_, err = database.DB.NamedExec(`UPDATE operationeventlog SET
"count"=:count
WHERE id=:id`, existingMessage)
if err != nil {
logging.LogError(err, "Failed to increase count on operationeventlog")
continue
}
if _, ok := unresolvedErrors[operationID][source][level]; !ok {
return 0
}
return unresolvedErrors[operationID][source][level]
}
func addUnresolvedError(operationID int, source string, level string, messageID int) {
if _, ok := unresolvedErrors[operationID]; !ok {
unresolvedErrors[operationID] = make(map[string]map[string]int)
}
if _, ok := unresolvedErrors[operationID][source]; !ok {
unresolvedErrors[operationID][source] = make(map[string]int)
}
unresolvedErrors[operationID][source][level] = messageID
}
func removeUnresolvedError(operationID int) {
delete(unresolvedErrors, operationID)
}
go RabbitMQConnection.EmitWebhookMessage(WebhookMessage{
OperationID: operation.ID,
OperationName: operation.Name,
OperationWebhook: operation.Webhook,
OperationChannel: operation.Channel,
OperatorUsername: "",
Action: WEBHOOK_TYPE_ALERT,
Data: map[string]interface{}{
"message": existingMessage.Message,
"source": existingMessage.Source,
"count": existingMessage.Count,
"timestamp": time.Now().UTC(),
},
})
} else {
newMessage := databaseStructs.Operationeventlog{
Source: sourceString,
Level: messageLevel,
Warning: warning,
Message: message,
OperationID: operation.ID,
Count: 1,
type operationMessage struct {
action string
message string
operationID int
source string
messageLevel database.MESSAGE_TYPE
warning bool
}
var operationMessageChannel = make(chan operationMessage, 100)
func listenForOperationsMessages() {
timer := time.NewTimer(time.Second * 10)
for {
select {
case msg := <-operationMessageChannel:
/*
Resolve a message in all operation's event logs if operationID is 0, otherwise just resolve it to the specific operation.
*/
var operations []databaseStructs.Operation
operationsMapMutex.RLock()
if msg.operationID == 0 {
for _, operation := range operationsMap {
operations = append(operations, operation)
}
if _, err := database.DB.NamedExec(`INSERT INTO operationeventlog
(source, "level", "message", operation_id, count, warning)
VALUES
(:source, :level, :message, :operation_id, :count, :warning)`, newMessage); err != nil {
logging.LogError(err, "Failed to create new operationeventlog message")
} else {
operation, ok := operationsMap[msg.operationID]
if !ok {
operationsMapMutex.RUnlock()
logging.LogError(nil, "unknown operation id, can't send message", "operation_id", msg.operationID)
continue
}
operations = append(operations, operation)
}
operationsMapMutex.RUnlock()
switch msg.action {
case "send":
sourceString := msg.source
if sourceString == "" {
sourceString = uuid.NewString()
}
// hitting concurrency issues where like-messages aren't getting collapsed like they should
var err error
for _, operation := range operations {
if msg.operationID == 0 || operation.ID == msg.operationID {
// this is the operation we're interested in
if msg.messageLevel == "warning" || msg.warning {
existingMessage := databaseStructs.Operationeventlog{}
existingMessage.ID = checkUnresolvedError(operation.ID, msg.source, msg.messageLevel)
if existingMessage.ID == 0 {
err = database.DB.Get(&existingMessage, `
SELECT id, count, "message", source, "level" FROM operationeventlog WHERE
warning=true and source=$1 and operation_id=$2 and resolved=false and deleted=false and "level"=$3
`, sourceString, operation.ID, msg.messageLevel)
if !errors.Is(err, sql.ErrNoRows) && err != nil {
logging.LogError(err, "Failed to query existing event log message")
continue
}
if errors.Is(err, sql.ErrNoRows) {
var utf8Message = strings.ToValidUTF8(string(bytes.ReplaceAll([]byte(msg.message), []byte{0}, []byte{})), "*")
if msg.messageLevel == "warning" {
msg.messageLevel = database.MESSAGE_LEVEL_INFO
}
newMessage := databaseStructs.Operationeventlog{
Source: sourceString,
Level: msg.messageLevel,
Warning: msg.warning,
Message: utf8Message,
OperationID: operation.ID,
Count: 1,
}
_, err = database.DB.NamedExec(`INSERT INTO operationeventlog
(source, "level", "message", operation_id, count, warning)
VALUES
(:source, :level, :message, :operation_id, :count, :warning)`, newMessage)
if err != nil {
logging.LogError(err, "Failed to create new operationeventlog message")
continue
}
go RabbitMQConnection.EmitWebhookMessage(WebhookMessage{
OperationID: operation.ID,
OperationName: operation.Name,
OperationWebhook: operation.Webhook,
OperationChannel: operation.Channel,
OperatorUsername: "",
Action: WEBHOOK_TYPE_ALERT,
Data: map[string]interface{}{
"message": newMessage.Message,
"source": newMessage.Source,
"count": newMessage.Count,
"timestamp": time.Now().UTC(),
},
})
EventingChannel <- EventNotification{
Trigger: eventing.TriggerAlert,
OperationID: operation.ID,
Outputs: map[string]interface{}{
"alert": newMessage.Message,
},
}
addUnresolvedError(operation.ID, msg.source, msg.messageLevel, existingMessage.ID)
continue
}
addUnresolvedError(operation.ID, msg.source, msg.messageLevel, existingMessage.ID)
}
if time.Now().Sub(lastUpdateTime).Seconds() > 1 {
lastUpdateTime = time.Now()
// don't bother trying to update if it's less than once a second
_, err = database.DB.NamedExec(`UPDATE operationeventlog SET
"count"="count" + 1 WHERE id=:id`, existingMessage)
if err != nil {
logging.LogError(err, "Failed to increase count on operationeventlog")
continue
}
}
/*
go RabbitMQConnection.EmitWebhookMessage(WebhookMessage{
OperationID: operation.ID,
OperationName: operation.Name,
OperationWebhook: operation.Webhook,
OperationChannel: operation.Channel,
OperatorUsername: "",
Action: WEBHOOK_TYPE_ALERT,
Data: map[string]interface{}{
"message": existingMessage.Message,
"source": existingMessage.Source,
"count": existingMessage.Count,
"timestamp": time.Now().UTC(),
},
})
*/
continue
}
var utf8Message = strings.ToValidUTF8(string(bytes.ReplaceAll([]byte(msg.message), []byte{0}, []byte{})), "*")
newMessage := databaseStructs.Operationeventlog{
Source: sourceString,
Level: msg.messageLevel,
Warning: msg.warning,
Message: utf8Message,
OperationID: operation.ID,
Count: 1,
}
if _, err := database.DB.NamedExec(`INSERT INTO operationeventlog
(source, "level", "message", operation_id, count, warning)
VALUES
(:source, :level, :message, :operation_id, :count, :warning)`, newMessage); err != nil {
logging.LogError(err, "Failed to create new operationeventlog message")
}
}
}
case "remove":
for _, operation := range operations {
if msg.operationID == 0 || operation.ID == msg.operationID {
// this is the operation we're interested in
updateObject := databaseStructs.Operationeventlog{
Message: msg.message,
OperationID: msg.operationID,
Level: database.MESSAGE_LEVEL_INFO,
}
if msg.operationID == 0 {
updateObject.OperationID = operation.ID
}
if _, err := database.DB.NamedExec(`UPDATE operationeventlog SET
resolved=true WHERE warning=true AND resolved=false AND deleted=false AND message=:message AND operation_id=:operation_id`, updateObject); err != nil {
logging.LogError(err, "Failed to resolve message")
}
removeUnresolvedError(operation.ID)
}
}
}
case <-timer.C:
if len(unresolvedErrors) > 0 {
unresolvedErrors = make(map[int]map[string]map[string]int)
}
timer.Reset(30 * time.Second)
}
}
}
func SendAllOperationsMessage(message string, operationID int, source string, messageLevel database.MESSAGE_TYPE, warning bool) {
operationMessageChannel <- operationMessage{
action: "send",
message: message,
source: source,
operationID: operationID,
messageLevel: messageLevel,
warning: warning,
}
}
func ResolveAllOperationsMessage(message string, operationID int) {
operationMessageChannel <- operationMessage{
action: "remove",
operationID: operationID,
message: message,
}
}
mythic-docker/src/utils/config.go
+1 -1
View File
@@ -11,7 +11,7 @@ import (
"github.com/spf13/viper"
)
const mythicServerVersion = "3.4.33"
const mythicServerVersion = "3.4.34"
type Config struct {
// server configuration
nginx-docker/Dockerfile
+1 -1
View File
@@ -6,5 +6,5 @@ RUN chmod +x /health_check.sh
RUN apk add openssl
RUN openssl req -x509 -newkey rsa:4096 -keyout /etc/ssl/private/mythic-ssl.key -out /etc/ssl/private/mythic-cert.crt -sha256 -days 3650 -nodes -subj "/O=Mythic/OU=Mythic/CN=Mythic"
HEALTHCHECK --interval=30s --timeout=60s --retries=5 --start-period=5s \
HEALTHCHECK --interval=30s --timeout=10s --retries=5 --start-period=5s \
CMD /health_check.sh || exit 1
postgres-docker/Dockerfile
+1 -1
View File
@@ -3,5 +3,5 @@ COPY ["pg_hba.conf", "/var/lib/postgresql/pg_hba.conf"]
COPY ["configuration.sh", "/docker-entrypoint-initdb.d/configuration.sh"]
COPY ["postgres.conf", "/etc/postgresql.conf"]
HEALTHCHECK --interval=30s --timeout=30s --retries=5 --start-period=5s \
HEALTHCHECK --interval=30s --timeout=10s --retries=5 --start-period=5s \
CMD pg_isready -d ${POSTGRES_DB:-mythic_db} -U ${POSTGRES_USER:-mythic_user} -p ${POSTGRES_PORT:-5432} || exit 1
rabbitmq-docker/Dockerfile
+1 -1
View File
@@ -3,7 +3,7 @@ ADD rabbitmq.conf /etc/rabbitmq/rabbitmq.conf
ADD generate_config.sh /generate_config.sh
ADD rabbitmq.conf /tmp/base_rabbitmq.conf
HEALTHCHECK --interval=60s --timeout=30s --retries=5 --start-period=5s \
HEALTHCHECK --interval=30s --timeout=10s --retries=5 --start-period=5s \
CMD rabbitmq-diagnostics -q check_port_connectivity || exit 1
CMD /bin/sh -c "chmod +x /generate_config.sh && /generate_config.sh && rabbitmq-server"