diff --git a/app.arc b/app.arc index 28dd2106b..5cf095b13 100644 --- a/app.arc +++ b/app.arc @@ -94,6 +94,16 @@ legacy_users email *String PointInTimeRecovery true +kafka_acls + topicName *String + group **String + PointInTimeRecovery true + +kafka_acl_log + partitionKey *Number + syncedOn **Number + PointInTimeRecovery ture + @tables-indexes email_notification_subscription topic *String diff --git a/app/lib/kafka.server.ts b/app/lib/kafka.server.ts index dcaa60ba0..5117d5bbc 100644 --- a/app/lib/kafka.server.ts +++ b/app/lib/kafka.server.ts @@ -5,13 +5,24 @@ * * SPDX-License-Identifier: Apache-2.0 */ +import { tables } from '@architect/functions' +import { paginateScan } from '@aws-sdk/lib-dynamodb' +import type { DynamoDBDocument } from '@aws-sdk/lib-dynamodb' import { Kafka } from 'gcn-kafka' +import type { AclEntry } from 'kafkajs' +import { + AclOperationTypes, + AclPermissionTypes, + AclResourceTypes, + ResourcePatternTypes, +} from 'kafkajs' import memoizee from 'memoizee' -import { domain, getEnvOrDie } from './env.server' +import { domain, getEnvOrDieInProduction } from './env.server' +import type { User } from '~/routes/_auth/user.server' -const client_id = getEnvOrDie('KAFKA_CLIENT_ID') -const client_secret = getEnvOrDie('KAFKA_CLIENT_SECRET') +const client_id = getEnvOrDieInProduction('KAFKA_CLIENT_ID') ?? '' +const client_secret = getEnvOrDieInProduction('KAFKA_CLIENT_SECRET') const kafka = new Kafka({ client_id, client_secret, @@ -68,3 +79,222 @@ if (process.env.ARC_SANDBOX) { await producer.send({ topic, messages: [{ value }] }) } } + +export type KafkaACL = { + topicName: string + permissionType: PermissionType + group: string + prefixed: boolean +} + +export type PermissionType = 'producer' | 'consumer' + +export const adminGroup = 'gcn.nasa.gov/gcn-admin' + +const consumerOperations = [AclOperationTypes.READ, AclOperationTypes.DESCRIBE] +const producerOperations = [ + AclOperationTypes.CREATE, + AclOperationTypes.WRITE, + AclOperationTypes.DESCRIBE, +] + +const admin_client_id = getEnvOrDieInProduction('KAFKA_ADMIN_CLIENT_ID') ?? '' +const admin_client_secret = getEnvOrDieInProduction('KAFKA_ADMIN_CLIENT_SECRET') +const adminKafka = new Kafka({ + client_id: admin_client_id, + client_secret: admin_client_secret, + domain, +}) + +function validateUser(user: User) { + if (!user.groups.includes(adminGroup)) + throw new Response(null, { status: 403 }) +} + +export async function createKafkaACL(user: User, acl: KafkaACL) { + validateUser(user) + // Save to db + const db = await tables() + await db.kafka_acls.put(acl) + + // Add to Kafka + const adminClient = adminKafka.admin() + await adminClient.connect() + await adminClient.createTopics({ + topics: [ + { + topic: acl.topicName, + }, + ], + }) + const acls = + acl.permissionType == 'producer' + ? createProducerAcls(acl) + : createConsumerAcls(acl) + await adminClient.createAcls({ acl: acls }) + await adminClient.disconnect() +} + +export async function getKafkaACLByTopicName(user: User, topicName: string) { + validateUser(user) + const db = await tables() + return (await db.kafka_acls.get({ topicName })) as KafkaACL +} + +export async function getKafkaACLsFromDynamoDB(user: User) { + validateUser(user) + const db = await tables() + const client = db._doc as unknown as DynamoDBDocument + const TableName = db.name('kafka_acls') + const pages = paginateScan({ client }, { TableName }) + const acls: KafkaACL[] = [] + for await (const page of pages) { + const newACL = page.Items as KafkaACL[] + if (newACL) acls.push(...newACL) + } + return acls +} + +export async function getAclsFromBrokers() { + const adminClient = adminKafka.admin() + await adminClient.connect() + const acls = await adminClient.describeAcls({ + resourceType: AclResourceTypes.TOPIC, + host: '*', + permissionType: AclPermissionTypes.ALLOW, + operation: AclOperationTypes.ANY, + resourcePatternType: ResourcePatternTypes.ANY, + }) + await adminClient.disconnect() + const results: KafkaACL[] = [] + for (const item of acls.resources) { + const topicName = item.resourceName + const prefixed = item.resourcePatternType === ResourcePatternTypes.PREFIXED + const producerRules = producerOperations.every((op) => + item.acls.map((x) => x.operation).includes(op) + ) + const producerGroup = + producerRules && + [ + ...new Set( + item.acls + .filter((acl) => producerOperations.includes(acl.operation)) + .map((x) => x.principal) + ), + ][0]?.replace('User:', '') + const consumerRules = consumerOperations.every((op) => + item.acls.map((x) => x.operation).includes(op) + ) + const consumerGroup = + consumerRules && + [ + ...new Set( + item.acls + .filter((acl) => consumerOperations.includes(acl.operation)) + .map((x) => x.principal) + ), + ][0]?.replace('User:', '') + if (producerRules && producerGroup) + results.push({ + topicName, + permissionType: 'producer', + group: producerGroup, + prefixed, + }) + if (consumerRules && consumerGroup) + results.push({ + topicName, + permissionType: 'consumer', + group: consumerGroup, + prefixed, + }) + } + return results +} + +export async function deleteKafkaACL(user: User, acl: KafkaACL) { + validateUser(user) + const db = await tables() + await db.kafka_acls.delete({ topicName: acl.topicName, group: acl.group }) + + const acls = + acl.permissionType == 'producer' + ? createProducerAcls(acl) + : createConsumerAcls(acl) + + const adminClient = adminKafka.admin() + await adminClient.connect() + await adminClient.deleteAcls({ filters: acls }) + await adminClient.disconnect() +} + +function createProducerAcls(acl: KafkaACL): AclEntry[] { + // Create, Write, and Describe operations + return mapAclAndOperations(acl, producerOperations) +} + +function createConsumerAcls(acl: KafkaACL): AclEntry[] { + // Read and Describe operations + return mapAclAndOperations(acl, consumerOperations) +} + +function mapAclAndOperations(acl: KafkaACL, operations: AclOperationTypes[]) { + return operations.map((operation) => { + return { + resourceType: AclResourceTypes.TOPIC, + resourceName: acl.topicName, + resourcePatternType: acl.prefixed + ? ResourcePatternTypes.PREFIXED + : ResourcePatternTypes.LITERAL, + principal: `User:${acl.group}`, + host: '*', + operation, + permissionType: AclPermissionTypes.ALLOW, + } + }) +} + +export async function updateBrokersFromDb(user: User) { + const dbDefinedAcls = await getKafkaACLsFromDynamoDB(user) + const mappedAcls = dbDefinedAcls.flatMap((x) => + x.permissionType === 'producer' + ? createProducerAcls(x) + : createConsumerAcls(x) + ) + + const adminClient = adminKafka.admin() + await adminClient.connect() + await adminClient.createAcls({ acl: mappedAcls }) + await adminClient.disconnect() +} + +export async function updateDbFromBrokers(user: User) { + const kafkaDefinedAcls = await getAclsFromBrokers() + const db = await tables() + await Promise.all([ + ...kafkaDefinedAcls.map((acl) => db.kafka_acls.put(acl)), + db.kafka_acl_log.put({ + partitionKey: 1, + syncedOn: Date.now(), + syncedBy: user.email, + }), + ]) +} + +type KafkaAclSyncLog = { + partitionKey: number + syncedOn: number + syncedBy: string +} + +export async function getLastSyncDate(): Promise { + const db = await tables() + return ( + await db.kafka_acl_log.query({ + KeyConditionExpression: 'partitionKey = :1', + ExpressionAttributeValues: { ':1': 1 }, + ScanIndexForward: false, + Limit: 1, + }) + ).Items.pop() as KafkaAclSyncLog +} diff --git a/app/root.tsx b/app/root.tsx index 60b43e296..2242d866b 100644 --- a/app/root.tsx +++ b/app/root.tsx @@ -46,6 +46,7 @@ import { useSpinDelay } from 'spin-delay' import invariant from 'tiny-invariant' import { features, getEnvOrDieInProduction, origin } from './lib/env.server' +import { adminGroup } from './lib/kafka.server' import { DevBanner } from './root/DevBanner' import { Footer } from './root/Footer' import NewsBanner from './root/NewsBanner' @@ -116,6 +117,7 @@ export async function loader({ request }: LoaderFunctionArgs) { const recaptchaSiteKey = getEnvOrDieInProduction('RECAPTCHA_SITE_KEY') const userIsMod = user?.groups.includes(moderatorGroup) const userIsVerifiedSubmitter = user?.groups.includes(group) + const userIsAdmin = user?.groups.includes(adminGroup) return { origin, @@ -126,6 +128,7 @@ export async function loader({ request }: LoaderFunctionArgs) { idp, userIsMod, userIsVerifiedSubmitter, + userIsAdmin, } } @@ -165,6 +168,11 @@ export function useSubmitterStatus() { return userIsVerifiedSubmitter } +export function useAdminStatus() { + const { userIsAdmin } = useLoaderDataRoot() + return userIsAdmin +} + export function useRecaptchaSiteKey() { const { recaptchaSiteKey } = useLoaderDataRoot() return recaptchaSiteKey diff --git a/app/root/header/Header.tsx b/app/root/header/Header.tsx index dde3947ad..61e3e9fcd 100644 --- a/app/root/header/Header.tsx +++ b/app/root/header/Header.tsx @@ -17,7 +17,7 @@ import { useEffect, useState } from 'react' import { useClickAnyWhere, useWindowSize } from 'usehooks-ts' import { Meatball } from '~/components/meatball/Meatball' -import { useEmail, useUserIdp } from '~/root' +import { useAdminStatus, useEmail, useUserIdp } from '~/root' import styles from './header.module.css' @@ -74,6 +74,7 @@ export function Header() { const [expanded, setExpanded] = useState(false) const [userMenuIsOpen, setUserMenuIsOpen] = useState(false) const isMobile = useWindowSize().width < 1024 + const userIsAdmin = useAdminStatus() function toggleMobileNav() { setExpanded((expanded) => !expanded) @@ -162,6 +163,11 @@ export function Header() { Profile , + userIsAdmin && ( + + Admin + + ), Peer Endorsements , diff --git a/app/routes/admin.kafka._index.tsx b/app/routes/admin.kafka._index.tsx new file mode 100644 index 000000000..0df4d7600 --- /dev/null +++ b/app/routes/admin.kafka._index.tsx @@ -0,0 +1,167 @@ +/*! + * Copyright © 2023 United States Government as represented by the + * Administrator of the National Aeronautics and Space Administration. + * All Rights Reserved. + * + * SPDX-License-Identifier: Apache-2.0 + */ +import type { LoaderFunctionArgs } from '@remix-run/node' +import { useFetcher, useLoaderData } from '@remix-run/react' +import type { ModalRef } from '@trussworks/react-uswds' +import { + Button, + Grid, + Icon, + Modal, + ModalFooter, + ModalHeading, + ModalToggleButton, +} from '@trussworks/react-uswds' +import { useRef } from 'react' + +import { getUser } from './_auth/user.server' +import HeadingWithAddButton from '~/components/HeadingWithAddButton' +import SegmentedCards from '~/components/SegmentedCards' +import Spinner from '~/components/Spinner' +import TimeAgo from '~/components/TimeAgo' +import type { KafkaACL } from '~/lib/kafka.server' +import { getKafkaACLsFromDynamoDB, getLastSyncDate } from '~/lib/kafka.server' + +export async function loader({ request }: LoaderFunctionArgs) { + const user = await getUser(request) + if (!user) throw new Response(null, { status: 403 }) + const dynamoDbAclData = await getKafkaACLsFromDynamoDB(user) + const latestSync = await getLastSyncDate() + return { dynamoDbAclData, latestSync } +} + +export default function Index() { + const { dynamoDbAclData, latestSync } = useLoaderData() + const aclFetcher = useFetcher() + + return ( + <> + Kafka Admin +

Kafka ACLs

+

+ Information about the Kafka ACLs listed here. Click the button to sync + the db to the kafka broker's current state. +

+ + + + {aclFetcher.state !== 'idle' && ( + + Saving... + + )} + + {latestSync && ( +

+ Last synced by {latestSync.syncedBy}{' '} + +

+ )} +

DynamoDB ACLs

+ {dynamoDbAclData && ( + <> + ({dynamoDbAclData.length}) ACLs + + {dynamoDbAclData + .sort((a, b) => a.topicName.localeCompare(b.topicName)) + .map((x, index) => ( + + ))} + + + )} + + ) +} + +function KafkaAclCard({ acl }: { acl: KafkaACL }) { + const ref = useRef(null) + const fetcher = useFetcher() + const disabled = fetcher.state !== 'idle' + + return ( + <> + +
+
+ + Topic: {acl.topicName} + +
+
+ + Permission Type: {acl.permissionType} + +
+
+ + Group: {acl.group} + +
+
+
+ + + Delete + +
+ + + + + + + + Delete Kafka ACL + + + + + Cancel + + + + + + + ) +} diff --git a/app/routes/admin.kafka.edit.tsx b/app/routes/admin.kafka.edit.tsx new file mode 100644 index 000000000..7d9af231a --- /dev/null +++ b/app/routes/admin.kafka.edit.tsx @@ -0,0 +1,95 @@ +/*! + * Copyright © 2023 United States Government as represented by the + * Administrator of the National Aeronautics and Space Administration. + * All Rights Reserved. + * + * SPDX-License-Identifier: Apache-2.0 + */ +import type { LoaderFunctionArgs } from '@remix-run/node' +import { Form, useLoaderData } from '@remix-run/react' +import { + Button, + Checkbox, + Label, + Select, + TextInput, +} from '@trussworks/react-uswds' + +import { getUser } from './_auth/user.server' +import { getGroups } from '~/lib/cognito.server' + +export async function loader({ request }: LoaderFunctionArgs) { + const user = await getUser(request) + if (!user) throw new Response(null, { status: 403 }) + const userGroups = (await getGroups()) + .filter((group) => group.GroupName?.startsWith('gcn.nasa.gov/')) + .map((group) => group.GroupName) + + return { userGroups } +} + +export default function Kafka() { + const { userGroups } = useLoaderData() + return +} + +function KafkaAclForm({ groups }: { groups: string[] }) { + return ( + <> +

Create Kafka ACLs

+
+ + console.log(e.target.value)} + /> + + + + Producer will generate ACLs for the Create, Write, and Describe + operations. Consumer will generate ACLs for the Read and Describe + operations + + + + +
+ + If yes, submission will also trigger th generation of ACLs for the + provided topic name as a PREFIXED topic with a period included at + the end. For example, if checked, a topic of `gcn.notices.icecube` + will result in ACLs for both `gcn.notices.icecube` (literal) and + `gcn.notices.icecube.` (prefixed). + +
+ + + + ) +} diff --git a/app/routes/admin.kafka.tsx b/app/routes/admin.kafka.tsx new file mode 100644 index 000000000..f42376550 --- /dev/null +++ b/app/routes/admin.kafka.tsx @@ -0,0 +1,101 @@ +/*! + * Copyright © 2023 United States Government as represented by the + * Administrator of the National Aeronautics and Space Administration. + * All Rights Reserved. + * + * SPDX-License-Identifier: Apache-2.0 + */ +import type { ActionFunctionArgs } from '@remix-run/node' +import { NavLink, Outlet } from '@remix-run/react' +import { GridContainer, SideNav } from '@trussworks/react-uswds' + +import { getUser } from './_auth/user.server' +import type { PermissionType } from '~/lib/kafka.server' +import { + adminGroup, + createKafkaACL, + deleteKafkaACL, + updateDbFromBrokers, +} from '~/lib/kafka.server' +import { getFormDataString } from '~/lib/utils' + +export async function action({ request }: ActionFunctionArgs) { + const user = await getUser(request) + if (!user?.groups.includes(adminGroup)) + throw new Response(null, { status: 403 }) + const data = await request.formData() + const intent = getFormDataString(data, 'intent') + if (intent === 'migrateFromBroker') { + await updateDbFromBrokers(user) + return null + } + const topicName = getFormDataString(data, 'topicName') + const permissionType = getFormDataString( + data, + 'permissionType' + ) as PermissionType + const group = getFormDataString(data, 'group') + const includePrefixed = getFormDataString(data, 'includePrefixed') + if (!topicName || !permissionType || !group) + throw new Response(null, { status: 400 }) + const promises = [] + + switch (intent) { + case 'delete': + promises.push( + deleteKafkaACL(user, { + topicName, + permissionType, + group, + prefixed: false, + }) + ) + break + case 'create': + promises.push( + createKafkaACL(user, { + topicName, + permissionType, + group, + prefixed: false, + }) + ) + + if (includePrefixed) + promises.push( + createKafkaACL(user, { + topicName: `${topicName}.`, + permissionType, + group, + prefixed: true, + }) + ) + break + default: + break + } + await Promise.all(promises) + + return null +} + +export default function Kafka() { + return ( + +
+
+ + Kafka + , + ]} + /> +
+
+ +
+
+ + ) +} diff --git a/playwright.config.ts b/playwright.config.ts index 5aa5401b4..76bf15d53 100644 --- a/playwright.config.ts +++ b/playwright.config.ts @@ -4,7 +4,6 @@ import { defineConfig, devices } from '@playwright/test' * Read environment variables from file. * https://github.com/motdotla/dotenv */ -// require('dotenv').config(); const deviceList = ['Desktop Firefox', 'Desktop Chrome', 'Desktop Safari'] @@ -69,7 +68,6 @@ export default defineConfig({ command: 'npm run dev', url: 'http://localhost:3333', reuseExistingServer: !process.env.CI, - stdout: 'pipe', timeout: 120 * 1000, // 120 Seconds timeout on webServer }, }) diff --git a/sandbox-seed.json b/sandbox-seed.json index 35534487e..64475fe61 100644 --- a/sandbox-seed.json +++ b/sandbox-seed.json @@ -5155,5 +5155,19 @@ "requestorEmail": "example@example.com", "subject": "Optical Observations for GRB 971227" } + ], + "kafka_acls": [ + { + "topicName": "test_topic_created_from_website", + "permissionType": "consumer", + "group": "gcn.nasa.gov/kafka-gcn-test-consumer", + "prefixed": false + }, + { + "topicName": "test_topic_created_from_website", + "permissionType": "producer", + "group": "gcn.nasa.gov/kafka-gcn-test-producer", + "prefixed": false + } ] }