fix(datasource): harden RSS connector and editor UX after review

Scope auth headers to feed fetches, use feed-scoped external IDs and
content-based incremental fingerprints, tolerate partial feed failures,
store feed URLs in settings, and fix credential configured detection.
This commit is contained in:
wizardchen
2026-06-23 15:45:47 +08:00
committed by lyingbug
parent e40ad802d7
commit 3cb0e906a7
12 changed files with 430 additions and 96 deletions
@@ -132,6 +132,21 @@ function syncRssAuthHeadersToCredentials() {
}
}
// Feed URLs may still live in credentials on older rows (not returned by the
// API). The backend copies them into settings on read; fall back to the
// selected feed resource IDs when settings are still empty.
function hydrateRssFeedUrlsFromConfig(config: { settings?: Record<string, any>; resource_ids?: string[] }) {
const settings = config.settings || {}
if (String(settings.feed_urls || '').trim()) {
return { ...settings }
}
const ids = config.resource_ids || []
if (ids.length === 0) {
return { ...settings }
}
return { ...settings, feed_urls: ids.join('\n') }
}
function addRssAuthHeader() {
rssAuthHeaders.value.push({ key: '', value: '' })
}
@@ -387,7 +402,6 @@ const connectorDefs = computed<ConnectorDef[]>(() => [
permissionPageUrl: '',
requiredPermissions: [],
fields: [
{ key: 'feed_urls', labelKey: 'datasource.field.feedUrls', placeholder: 'https://example.com/feed.xml', multiline: true, hintKey: 'datasource.field.feedUrlsHint' },
{ key: 'auth_headers', labelKey: 'datasource.field.authHeaders', placeholder: '', optional: true, hintKey: 'datasource.field.authHeadersHint', fieldType: 'custom_headers' },
],
},
@@ -431,13 +445,16 @@ watch(visible, async (v) => {
credentialsConfigured.value = false
refreshCredentialsStatus()
testResult.value = credentialsConfigured.value ? 'success' : ''
const editConfig = props.dataSource.config || {}
form.value = {
name: props.dataSource.name,
type: props.dataSource.type,
config: {
credentials: {},
resource_ids: props.dataSource.config?.resource_ids || [],
settings: props.dataSource.config?.settings || {},
resource_ids: editConfig.resource_ids || [],
settings: props.dataSource.type === 'rss'
? hydrateRssFeedUrlsFromConfig(editConfig)
: (editConfig.settings || {}),
},
sync_schedule: props.dataSource.sync_schedule,
sync_mode: props.dataSource.sync_mode,
@@ -484,6 +501,16 @@ watch(
{ deep: true },
)
watch(
() => form.value.config.settings.feed_urls,
() => {
if (needsConnectionTest()) {
testResult.value = ''
testErrorMsg.value = ''
}
},
)
function selectType(def: ConnectorDef) {
if (!def.available) return
form.value.type = def.type
@@ -496,6 +523,7 @@ function selectType(def: ConnectorDef) {
// --- Test connection (stateless, no DB write) ---
async function testConnection() {
syncRssAuthHeadersToCredentials()
if (!validateRssFeedUrls()) return
if (!isEdit.value || !credentialsConfigured.value || replaceCredentialsMode.value) {
const fields = currentDef.value?.fields || []
for (const f of fields) {
@@ -518,7 +546,12 @@ async function testConnection() {
} as any)
await validateConnection(tempDsId.value)
} else {
await validateCredentials(form.value.type, form.value.config.credentials)
const creds = { ...form.value.config.credentials }
if (form.value.type === 'rss') {
// validate-credentials is credentials-only; feed URLs live in settings.
creds.feed_urls = form.value.config.settings.feed_urls
}
await validateCredentials(form.value.type, creds)
}
testResult.value = 'success'
MessagePlugin.success(t('datasource.testSuccess'))
@@ -672,8 +705,18 @@ function toggleResource(id: string) {
selectedResourceIds.value = [...cover]
}
function validateRssFeedUrls(): boolean {
if (form.value.type !== 'rss') return true
if (!String(form.value.config.settings.feed_urls || '').trim()) {
MessagePlugin.warning(`${t('datasource.field.feedUrls')} ${t('datasource.isRequired')}`)
return false
}
return true
}
function validateStep1Fields(): boolean {
syncRssAuthHeadersToCredentials()
if (!validateRssFeedUrls()) return false
if (isEdit.value && credentialsConfigured.value && !replaceCredentialsMode.value) {
return true
}
@@ -1079,6 +1122,21 @@ const drawerConfirmText = computed(() => {
</div>
</section>
<section v-if="form.type === 'rss'" class="setting-drawer__section">
<h4 class="setting-drawer__section-title">{{ t('datasource.field.feedUrls') }}</h4>
<div class="form-item">
<label class="form-label required">{{ t('datasource.field.feedUrls') }}</label>
<t-textarea
v-model="form.config.settings.feed_urls"
placeholder="https://example.com/feed.xml"
:autosize="{ minRows: 2, maxRows: 6 }"
autocomplete="off"
spellcheck="false"
/>
<p class="form-desc">{{ t('datasource.field.feedUrlsHint') }}</p>
</div>
</section>
<section class="setting-drawer__section">
<h4 class="setting-drawer__section-title">{{ t('datasource.credentialsLabel') }}</h4>
@@ -80,6 +80,12 @@ func (s *DataSourceService) CreateDataSource(ctx context.Context, ds *types.Data
}
// Validate configuration
if cfg, err := ds.ParseConfig(); err == nil && cfg != nil {
cfg.StripNonSecretCredentials(ds.Type)
if blob, err := cfg.ToJSON(); err == nil {
ds.Config = blob
}
}
if err := s.validateDataSourceConfig(ctx, ds); err != nil {
return nil, err
}
@@ -176,6 +182,7 @@ func (s *DataSourceService) UpdateDataSource(ctx context.Context, ds *types.Data
} else {
merged.Credentials = nil
}
merged.StripNonSecretCredentials(ds.Type)
if blob, err := merged.ToJSON(); err == nil {
ds.Config = blob
}
@@ -192,7 +199,7 @@ func (s *DataSourceService) UpdateDataSource(ctx context.Context, ds *types.Data
if mergedCfg != nil && existingParsedCfg != nil {
configActuallyChanged = !reflect.DeepEqual(*mergedCfg, *existingParsedCfg)
}
hasCreds := mergedCfg != nil && mergedCfg.HasCredentials()
hasCreds := mergedCfg != nil && mergedCfg.HasConfiguredCredentials(ds.Type)
if hasCreds && (ds.Type != existing.Type || configActuallyChanged) {
if err := s.validateDataSourceConfig(ctx, ds); err != nil {
return nil, err
@@ -236,6 +243,7 @@ func (s *DataSourceService) UpdateDataSourceCredentials(
parsed = &types.DataSourceConfig{Type: existing.Type}
}
parsed.Credentials = credentials
parsed.StripNonSecretCredentials(existing.Type)
blob, err := parsed.ToJSON()
if err != nil {
return nil, err
@@ -269,9 +277,18 @@ func (s *DataSourceService) ClearDataSourceCredentials(ctx context.Context, id s
if err != nil {
return err
}
if parsed == nil || !parsed.HasCredentials() {
if parsed == nil {
return nil
}
parsed.StripNonSecretCredentials(existing.Type)
if !parsed.HasConfiguredCredentials(existing.Type) {
blob, err := parsed.ToJSON()
if err != nil {
return err
}
existing.Config = blob
return s.dsRepo.Update(ctx, existing)
}
parsed.Credentials = nil
blob, err := parsed.ToJSON()
if err != nil {
@@ -562,6 +579,16 @@ func (s *DataSourceService) ProcessSync(ctx context.Context, task *asynq.Task) e
return nil
}
if _, err := s.kbService.GetKnowledgeBaseByID(ctx, ds.KnowledgeBaseID); err != nil {
logger.Warnf(ctx, "knowledge base not found (likely deleted), cancelling sync: kb=%s ds=%s err=%v",
ds.KnowledgeBaseID, payload.DataSourceID, err)
syncLog.Status = types.SyncLogStatusCanceled
syncLog.FinishedAt = timePtr(time.Now().UTC())
syncLog.ErrorMessage = "knowledge base has been deleted"
_ = s.syncLogRepo.Update(ctx, syncLog)
return nil
}
wasPaused := ds.Status == types.DataSourceStatusPaused
// Get connector
+2 -2
View File
@@ -200,8 +200,8 @@ var ConnectorMetadataRegistry = map[string]ConnectorMetadata{
Name: "RSS / Atom Feed",
Description: "Sync articles from RSS/Atom feeds",
Priority: 12,
AuthType: "none",
Capabilities: []string{},
AuthType: "custom",
Capabilities: []string{"incremental"},
},
}
+10 -8
View File
@@ -42,10 +42,10 @@ func newClient(headers map[string]string) *client {
}
}
// fetch retrieves rawURL with SSRF validation and size limiting. The user's
// custom headers are applied first, then a default User-Agent / Accept if the
// user did not override them.
func (c *client) fetch(ctx context.Context, rawURL string, maxSize int64) ([]byte, error) {
// fetch retrieves rawURL with SSRF validation and size limiting. Custom auth
// headers are only attached when withAuthHeaders is true (feed fetches); article
// pages on third-party domains must not receive feed credentials.
func (c *client) fetch(ctx context.Context, rawURL string, maxSize int64, withAuthHeaders bool) ([]byte, error) {
if err := utils.ValidateURLForSSRF(rawURL); err != nil {
return nil, fmt.Errorf("URL rejected: %w", err)
}
@@ -61,8 +61,10 @@ func (c *client) fetch(ctx context.Context, rawURL string, maxSize int64) ([]byt
return nil, err
}
for k, v := range c.headers {
req.Header.Set(k, v)
if withAuthHeaders {
for k, v := range c.headers {
req.Header.Set(k, v)
}
}
if req.Header.Get("User-Agent") == "" {
req.Header.Set("User-Agent", defaultUserAgent)
@@ -91,7 +93,7 @@ func (c *client) fetch(ctx context.Context, rawURL string, maxSize int64) ([]byt
// fetchFeed retrieves the raw bytes of a feed document.
func (c *client) fetchFeed(ctx context.Context, feedURL string) ([]byte, error) {
return c.fetch(ctx, feedURL, maxFeedSize)
return c.fetch(ctx, feedURL, maxFeedSize, true)
}
// extractArticle fetches an article page and returns the readability-cleaned
@@ -99,7 +101,7 @@ func (c *client) fetchFeed(ctx context.Context, feedURL string) ([]byte, error)
// error if the page can't be fetched or no readable content is found, so the
// caller can fall back to feed-provided content.
func (c *client) extractArticle(ctx context.Context, articleURL string) (contentHTML, title string, err error) {
body, err := c.fetch(ctx, articleURL, maxArticleSize)
body, err := c.fetch(ctx, articleURL, maxArticleSize, false)
if err != nil {
return "", "", err
}
+63 -39
View File
@@ -124,18 +124,23 @@ func (c *Connector) FetchAll(
return items, err
}
// FetchIncremental returns only items whose fingerprint changed since the prior
// cursor. Deletions are intentionally not emitted (feeds drop old items as a
// matter of course).
// FetchIncremental returns only items whose content fingerprint changed since
// the prior cursor. Deletions are intentionally not emitted (feeds drop old
// items as a matter of course).
func (c *Connector) FetchIncremental(
ctx context.Context, config *types.DataSourceConfig, cursor *types.SyncCursor,
) ([]types.FetchedItem, *types.SyncCursor, error) {
var prev *rssCursor
if cursor != nil && cursor.ConnectorCursor != nil {
var p rssCursor
b, _ := json.Marshal(cursor.ConnectorCursor)
_ = json.Unmarshal(b, &p)
prev = &p
b, err := json.Marshal(cursor.ConnectorCursor)
if err != nil {
logger.Warnf(ctx, "[RSS] marshal connector cursor: %v", err)
} else if err := json.Unmarshal(b, &p); err != nil {
logger.Warnf(ctx, "[RSS] unmarshal connector cursor: %v", err)
} else {
prev = &p
}
}
items, newCursor, err := c.walk(ctx, config, config.ResourceIDs, prev, true)
@@ -144,8 +149,12 @@ func (c *Connector) FetchIncremental(
}
cursorMap := make(map[string]interface{})
b, _ := json.Marshal(newCursor)
_ = json.Unmarshal(b, &cursorMap)
b, err := json.Marshal(newCursor)
if err != nil {
logger.Warnf(ctx, "[RSS] marshal new cursor: %v", err)
} else if err := json.Unmarshal(b, &cursorMap); err != nil {
logger.Warnf(ctx, "[RSS] unmarshal new cursor to map: %v", err)
}
return items, &types.SyncCursor{
LastSyncTime: newCursor.LastSyncTime,
@@ -155,9 +164,9 @@ func (c *Connector) FetchIncremental(
// walk is the shared implementation for FetchAll / FetchIncremental.
//
// When incremental is true, items whose fingerprint matches the prior cursor
// are skipped (no article re-fetch). The returned cursor always reflects the
// full current item set so a later sync can detect changes.
// When incremental is true, items whose content fingerprint matches the prior
// cursor are omitted from the result (ingest skipped). The returned cursor
// always reflects the full current item set so a later sync can detect changes.
func (c *Connector) walk(
ctx context.Context,
config *types.DataSourceConfig,
@@ -184,15 +193,22 @@ func (c *Connector) walk(
FeedItems: make(map[string]map[string]string),
}
var out []types.FetchedItem
var feedErrors []string
for _, feedURL := range feedURLs {
data, err := cli.fetchFeed(ctx, feedURL)
if err != nil {
return nil, nil, fmt.Errorf("fetch feed %s: %w", feedURL, err)
logger.Warnf(ctx, "[RSS] sync: fetch %s failed: %v", feedURL, err)
feedErrors = append(feedErrors, fmt.Sprintf("%s: %v", feedURL, err))
newCursor.FeedItems[feedURL] = make(map[string]string)
continue
}
feed, err := parser.Parse(bytes.NewReader(data))
if err != nil {
return nil, nil, fmt.Errorf("parse feed %s: %w", feedURL, err)
logger.Warnf(ctx, "[RSS] sync: parse %s failed: %v", feedURL, err)
feedErrors = append(feedErrors, fmt.Sprintf("%s: %v", feedURL, err))
newCursor.FeedItems[feedURL] = make(map[string]string)
continue
}
newCursor.FeedItems[feedURL] = make(map[string]string)
@@ -212,38 +228,43 @@ func (c *Connector) walk(
}
feedContent := firstNonEmpty(item.Content, item.Description)
fp := itemFingerprint(item.UpdatedParsed, item.PublishedParsed, feedContent)
newCursor.FeedItems[feedURL][itemID] = fp
resolved := c.resolveItem(ctx, cli, feed, item, feedURL, itemID, feedContent)
newCursor.FeedItems[feedURL][itemID] = resolved.fingerprint
if incremental && prevItems != nil && prevItems[itemID] == fp {
if incremental && prevItems != nil && prevItems[itemID] == resolved.fingerprint {
skipped++
continue
}
kept++
out = append(out, c.buildItem(ctx, cli, feed, item, feedURL, itemID, feedContent))
out = append(out, resolved.item)
}
logger.Infof(ctx, "[RSS] feed %s: items=%d fetched=%d skipped=%d",
feedURL, len(feed.Items), kept, skipped)
}
if !incremental {
return out, newCursor, nil
if len(out) == 0 && len(feedErrors) > 0 && len(feedErrors) == len(feedURLs) {
return nil, nil, fmt.Errorf("all feeds failed: %s", strings.Join(feedErrors, "; "))
}
return out, newCursor, nil
}
// buildItem assembles a FetchedItem for a single feed entry, resolving the
type resolvedFeedItem struct {
item types.FetchedItem
fingerprint string
}
// resolveItem assembles a FetchedItem for a single feed entry, resolving the
// best available content (full-text article > feed content) and converting it
// to Markdown.
func (c *Connector) buildItem(
func (c *Connector) resolveItem(
ctx context.Context,
cli *client,
feed *gofeed.Feed,
item *gofeed.Item,
feedURL, itemID, feedContent string,
) types.FetchedItem {
) resolvedFeedItem {
title := firstNonEmpty(item.Title, "untitled")
// Prefer full article text; fall back to feed-provided content on failure.
@@ -274,22 +295,25 @@ func (c *Connector) buildItem(
author = item.Author.Name
}
return types.FetchedItem{
ExternalID: itemID,
Title: title,
Content: []byte(content),
ContentType: "text/markdown",
FileName: sanitizeFileName(title) + ".md",
URL: item.Link,
UpdatedAt: updatedAt,
SourceResourceID: feedURL,
Metadata: map[string]string{
"channel": types.ChannelRSS,
"feed_url": feedURL,
"feed_title": feed.Title,
"guid": item.GUID,
"link": item.Link,
"author": author,
return resolvedFeedItem{
fingerprint: contentFingerprint(content),
item: types.FetchedItem{
ExternalID: itemExternalID(feedURL, itemID),
Title: title,
Content: []byte(content),
ContentType: "text/markdown",
FileName: sanitizeFileName(title) + ".md",
URL: item.Link,
UpdatedAt: updatedAt,
SourceResourceID: feedURL,
Metadata: map[string]string{
"channel": types.ChannelRSS,
"feed_url": feedURL,
"feed_title": feed.Title,
"guid": item.GUID,
"link": item.Link,
"author": author,
},
},
}
}
@@ -32,9 +32,10 @@ const longArticleBody = `<p>This is the first paragraph of a reasonably long art
// fakeFeed spins up an httptest server serving an RSS feed and article pages.
type fakeFeed struct {
server *httptest.Server
feedTitle string
itemContent string // optional <description>/content for items
server *httptest.Server
feedTitle string
itemContent string // optional <description>/content for items
articleAuthHeaders []string
}
func newFakeFeed(t *testing.T) *fakeFeed {
@@ -43,6 +44,11 @@ func newFakeFeed(t *testing.T) *fakeFeed {
mux := http.NewServeMux()
mux.HandleFunc("/article/", func(w http.ResponseWriter, r *http.Request) {
for k, vals := range r.Header {
if strings.EqualFold(k, "X-Test-Auth") && len(vals) > 0 && vals[0] != "" {
f.articleAuthHeaders = append(f.articleAuthHeaders, vals[0])
}
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
fmt.Fprintf(w, `<!DOCTYPE html><html><head><title>%s</title></head>`+
`<body><nav>menu</nav><article><h1>Heading</h1>%s</article><footer>foot</footer></body></html>`,
@@ -87,14 +93,17 @@ func newFakeFeed(t *testing.T) *fakeFeed {
func (f *fakeFeed) feedURL() string { return f.server.URL + "/feed.xml" }
func makeConfig(feedURLs string, headers string) *types.DataSourceConfig {
creds := map[string]interface{}{"feed_urls": feedURLs}
cfg := &types.DataSourceConfig{
Type: types.ConnectorTypeRSS,
Settings: map[string]interface{}{
"feed_urls": feedURLs,
},
Credentials: map[string]interface{}{},
}
if headers != "" {
creds["auth_headers"] = headers
}
return &types.DataSourceConfig{
Type: types.ConnectorTypeRSS,
Credentials: creds,
cfg.Credentials["auth_headers"] = headers
}
return cfg
}
func TestConnector_Type(t *testing.T) {
@@ -109,6 +118,33 @@ func TestParseConfig_RequiresFeedURLs(t *testing.T) {
}
}
func TestParseConfig_FeedURLsFromSettings(t *testing.T) {
cfg, err := parseConfig(makeConfig("https://example.com/feed.xml", ""))
if err != nil {
t.Fatalf("parseConfig error: %v", err)
}
got := cfg.feedURLList()
if len(got) != 1 || got[0] != "https://example.com/feed.xml" {
t.Fatalf("feedURLList = %v", got)
}
}
func TestParseConfig_LegacyFeedURLsInCredentials(t *testing.T) {
legacy := &types.DataSourceConfig{
Type: types.ConnectorTypeRSS,
Credentials: map[string]interface{}{
"feed_urls": "https://legacy.example/feed.xml",
},
}
cfg, err := parseConfig(legacy)
if err != nil {
t.Fatalf("parseConfig error: %v", err)
}
if got := cfg.feedURLList(); len(got) != 1 || got[0] != "https://legacy.example/feed.xml" {
t.Fatalf("feedURLList = %v", got)
}
}
func TestConfig_FeedURLList_SplitsAndDedupes(t *testing.T) {
cfg := &Config{FeedURLs: "https://a.com/f, https://b.com/f\nhttps://a.com/f\n"}
got := cfg.feedURLList()
@@ -132,6 +168,14 @@ func TestConfig_ParseHeaders(t *testing.T) {
}
}
func TestItemExternalID_ScopesByFeed(t *testing.T) {
got := itemExternalID("https://a.com/feed", "guid-1")
want := "https://a.com/feed:guid-1"
if got != want {
t.Fatalf("itemExternalID = %q, want %q", got, want)
}
}
func TestConnector_Validate_Success(t *testing.T) {
f := newFakeFeed(t)
if err := NewConnector().Validate(context.Background(), makeConfig(f.feedURL(), "")); err != nil {
@@ -155,6 +199,20 @@ func TestConnector_Validate_PrivateFeedWithHeader(t *testing.T) {
}
}
func TestConnector_FetchAll_DoesNotSendAuthHeadersToArticles(t *testing.T) {
f := newFakeFeed(t)
f.itemContent = "needs-auth"
_, err := NewConnector().FetchAll(
context.Background(), makeConfig(f.feedURL(), "X-Test-Auth: secret"), nil,
)
if err != nil {
t.Fatalf("FetchAll error: %v", err)
}
if len(f.articleAuthHeaders) != 0 {
t.Fatalf("article requests must not carry feed auth headers, got %v", f.articleAuthHeaders)
}
}
func TestConnector_ListResources(t *testing.T) {
f := newFakeFeed(t)
res, err := NewConnector().ListResources(context.Background(), makeConfig(f.feedURL(), ""), "")
@@ -192,8 +250,9 @@ func TestConnector_FetchAll_FullTextMarkdown(t *testing.T) {
}
it := items[0]
if it.ExternalID != "guid-1" {
t.Errorf("ExternalID = %q, want guid-1", it.ExternalID)
wantExternalID := itemExternalID(f.feedURL(), "guid-1")
if it.ExternalID != wantExternalID {
t.Errorf("ExternalID = %q, want %q", it.ExternalID, wantExternalID)
}
if it.ContentType != "text/markdown" {
t.Errorf("ContentType = %q, want text/markdown", it.ContentType)
@@ -237,6 +296,18 @@ func TestConnector_FetchIncremental_SkipsUnchanged(t *testing.T) {
}
}
func TestConnector_Walk_PartialFeedFailure(t *testing.T) {
f := newFakeFeed(t)
cfg := makeConfig(f.feedURL()+", https://invalid.invalid/feed.xml", "")
items, err := NewConnector().FetchAll(context.Background(), cfg, nil)
if err != nil {
t.Fatalf("expected partial success, got error: %v", err)
}
if len(items) != 2 {
t.Fatalf("expected 2 items from healthy feed, got %d", len(items))
}
}
func TestConnector_ResolveResourceAncestors_Empty(t *testing.T) {
got, err := NewConnector().ResolveResourceAncestors(
context.Background(), makeConfig("https://a.com/f", ""), []string{"x"},
+40 -27
View File
@@ -7,15 +7,15 @@
//
// Capabilities:
// - Private feeds: optional custom HTTP headers (e.g. "Authorization: Bearer …")
// are attached to every request, both to the feed and to article pages.
// are attached only to feed fetches, never to third-party article pages.
// - Full text: when an item exposes a link, the article page is fetched and
// run through a readability extractor; the cleaned HTML is converted to
// Markdown. Feed-provided content (content:encoded / description) is used
// as a fallback.
// - Incremental: a per-feed fingerprint (item updated/published time, or a
// content hash when no timestamp is present) detects changed items so only
// new/updated articles are re-fetched. Deletions are NOT synced — feeds
// routinely drop old items, which would otherwise look like deletions.
// - Incremental: a per-item fingerprint hashes the final Markdown body that
// would be ingested, so feed-only metadata changes and full-article edits
// are both detected. Deletions are NOT synced — feeds routinely drop old
// items, which would otherwise look like deletions.
//
// All outbound requests go through the SSRF-safe HTTP client so a malicious
// feed cannot redirect WeKnora to internal services.
@@ -34,22 +34,22 @@ import (
"github.com/Tencent/WeKnora/internal/types"
)
// Config holds RSS-specific configuration parsed from DataSourceConfig.Credentials.
// Config holds RSS-specific configuration.
//
// Both fields live under Credentials (rather than Settings) because the
// existing data-source editor UI only renders credential inputs, and because
// AuthHeaders may carry secrets that must be encrypted at rest. FeedURLs are
// not secret but ride along for the same single-input flow.
// FeedURLs are stored in DataSourceConfig.Settings (non-secret, editable in
// the UI without replacing credentials). AuthHeaders live in Credentials
// because they may carry secrets that must be encrypted at rest. Credentials
// may still carry feed_urls for backward compatibility with older rows.
type Config struct {
// FeedURLs is a newline- or comma-separated list of feed URLs.
FeedURLs string `json:"feed_urls"`
// AuthHeaders is an optional newline-separated list of custom request
// headers in "Name: Value" form, applied to every outbound request.
// headers in "Name: Value" form, applied only to feed fetches.
AuthHeaders string `json:"auth_headers,omitempty"`
}
// parseConfig extracts and validates RSS configuration from the credentials map.
// parseConfig extracts and validates RSS configuration.
func parseConfig(config *types.DataSourceConfig) (*Config, error) {
if config == nil {
return nil, fmt.Errorf("%w: config is nil", datasource.ErrInvalidConfig)
@@ -62,12 +62,30 @@ func parseConfig(config *types.DataSourceConfig) (*Config, error) {
if err := json.Unmarshal(credBytes, &cfg); err != nil {
return nil, fmt.Errorf("parse rss credentials: %w", err)
}
if urls := feedURLsFromSettings(config.Settings); urls != "" {
cfg.FeedURLs = urls
}
if len(cfg.feedURLList()) == 0 {
return nil, fmt.Errorf("%w: feed_urls is required", datasource.ErrInvalidCredentials)
}
return &cfg, nil
}
func feedURLsFromSettings(settings map[string]interface{}) string {
if len(settings) == 0 {
return ""
}
raw, ok := settings["feed_urls"]
if !ok {
return ""
}
s, ok := raw.(string)
if !ok {
return ""
}
return strings.TrimSpace(s)
}
// feedURLList splits FeedURLs on newlines and commas, trims, dedupes (order
// preserved), and drops blanks.
func (c *Config) feedURLList() []string {
@@ -124,29 +142,24 @@ func (c *Config) parseHeaders() map[string]string {
// rssCursor stores incremental sync state.
//
// FeedItems maps feedURL → itemID → fingerprint, where fingerprint is the
// item's updated/published timestamp (RFC3339) or, when no timestamp exists,
// a "h:<sha256-prefix>" content hash. An item is considered unchanged when its
// fingerprint matches the stored one.
// FeedItems maps feedURL → itemID → fingerprint, where fingerprint is a
// "h:<sha256-prefix>" hash of the final Markdown body that would be ingested.
type rssCursor struct {
LastSyncTime time.Time `json:"last_sync_time"`
FeedItems map[string]map[string]string `json:"feed_items,omitempty"`
}
// itemFingerprint returns a change-detection token for a feed item. It prefers
// the explicit updated/published timestamp; absent that, it hashes the content
// so edits to timestamp-less items are still detected.
func itemFingerprint(updated, published *time.Time, content string) string {
if updated != nil && !updated.IsZero() {
return updated.UTC().Format(time.RFC3339)
}
if published != nil && !published.IsZero() {
return published.UTC().Format(time.RFC3339)
}
sum := sha256.Sum256([]byte(content))
// contentFingerprint hashes ingested Markdown for incremental change detection.
func contentFingerprint(markdown string) string {
sum := sha256.Sum256([]byte(markdown))
return "h:" + hex.EncodeToString(sum[:])[:16]
}
// itemExternalID scopes an item ID to its feed so GUIDs cannot collide across feeds.
func itemExternalID(feedURL, itemID string) string {
return feedURL + ":" + itemID
}
// firstNonEmpty returns the first non-empty trimmed string among the args.
func firstNonEmpty(values ...string) string {
for _, v := range values {
+2 -2
View File
@@ -83,8 +83,8 @@ func (h *DataSourceCredentialsHandler) Put(c *gin.Context) {
return
}
configured := false
if parsed, err := updated.ParseConfig(); err == nil && parsed != nil && parsed.HasCredentials() {
configured = true
if parsed, err := updated.ParseConfig(); err == nil && parsed != nil {
configured = parsed.HasConfiguredCredentials(updated.Type)
}
resp := dto.CredentialsResponse{
Fields: map[string]dto.CredentialFieldMetadata{
+29 -1
View File
@@ -2,6 +2,7 @@ package dto
import (
"encoding/json"
"strings"
"time"
"github.com/Tencent/WeKnora/internal/types"
@@ -65,7 +66,8 @@ func NewDataSourceResponse(ds *types.DataSource) *DataSourceResponse {
ResourceIDs: parsed.ResourceIDs,
Settings: parsed.Settings,
}
configured = parsed.HasCredentials()
enrichRSSFeedURLsInSettings(ds.Type, parsed, cfgDTO)
configured = parsed.HasConfiguredCredentials(ds.Type)
}
return &DataSourceResponse{
ID: ds.ID,
@@ -94,6 +96,32 @@ func NewDataSourceResponse(ds *types.DataSource) *DataSourceResponse {
}
}
// enrichRSSFeedURLsInSettings copies feed_urls from credentials into settings
// for API responses. Feed URLs are not secrets but may still live in the
// encrypted credentials blob on rows created before they moved to settings.
func enrichRSSFeedURLsInSettings(dsType string, parsed *types.DataSourceConfig, cfgDTO *DataSourceConfigDTO) {
if dsType != types.ConnectorTypeRSS || parsed == nil || cfgDTO == nil {
return
}
if cfgDTO.Settings != nil {
if v, ok := cfgDTO.Settings["feed_urls"].(string); ok && strings.TrimSpace(v) != "" {
return
}
}
raw, ok := parsed.Credentials["feed_urls"]
if !ok {
return
}
feedURLs, ok := raw.(string)
if !ok || strings.TrimSpace(feedURLs) == "" {
return
}
if cfgDTO.Settings == nil {
cfgDTO.Settings = make(map[string]interface{})
}
cfgDTO.Settings["feed_urls"] = feedURLs
}
func NewDataSourceResponses(dss []*types.DataSource) []*DataSourceResponse {
out := make([]*DataSourceResponse, 0, len(dss))
for _, d := range dss {
+50
View File
@@ -53,6 +53,56 @@ func TestDataSourceResponse_NilSafe(t *testing.T) {
assert.Equal(t, []*DataSourceResponse{}, NewDataSourceResponses(nil))
}
func TestDataSourceResponse_RSSFeedURLsFromCredentials(t *testing.T) {
cfg := types.DataSourceConfig{
Type: types.ConnectorTypeRSS,
Credentials: map[string]interface{}{
"feed_urls": "https://example.com/a.xml\nhttps://example.com/b.xml",
},
ResourceIDs: []string{"https://example.com/a.xml"},
}
blob, _ := cfg.ToJSON()
ds := &types.DataSource{
ID: "ds-rss",
Name: "my-rss",
Type: types.ConnectorTypeRSS,
Config: blob,
}
resp := NewDataSourceResponse(ds)
assert.NotNil(t, resp.Config)
assert.Equal(t, "https://example.com/a.xml\nhttps://example.com/b.xml", resp.Config.Settings["feed_urls"])
assert.False(t, resp.Credentials["credentials"].Configured)
body, err := json.Marshal(resp)
assert.NoError(t, err)
s := string(body)
assert.Contains(t, s, "https://example.com/a.xml")
assert.NotContains(t, s, "Bearer secret")
}
func TestDataSourceResponse_RSSAuthHeadersConfigured(t *testing.T) {
cfg := types.DataSourceConfig{
Type: types.ConnectorTypeRSS,
Credentials: map[string]interface{}{
"auth_headers": "Authorization: Bearer secret",
},
Settings: map[string]interface{}{
"feed_urls": "https://example.com/feed.xml",
},
}
blob, _ := cfg.ToJSON()
ds := &types.DataSource{
ID: "ds-rss-auth",
Name: "my-rss",
Type: types.ConnectorTypeRSS,
Config: blob,
}
resp := NewDataSourceResponse(ds)
assert.True(t, resp.Credentials["credentials"].Configured)
body, err := json.Marshal(resp)
assert.NoError(t, err)
assert.NotContains(t, string(body), "Bearer secret")
}
func TestDataSourceResponse_NoConfig(t *testing.T) {
ds := &types.DataSource{ID: "x", Name: "x"}
body, err := json.Marshal(NewDataSourceResponse(ds))
+36
View File
@@ -3,6 +3,7 @@ package types
import (
"encoding/json"
"log"
"strings"
"time"
"github.com/Tencent/WeKnora/internal/utils"
@@ -223,6 +224,41 @@ func (d DataSourceConfig) HasCredentials() bool {
return len(d.Credentials) > 0
}
// HasConfiguredCredentials reports whether user-facing secret credentials are
// stored. RSS feed URLs are non-secret configuration (settings); only
// auth_headers count as credentials for that connector.
func (d DataSourceConfig) HasConfiguredCredentials(connectorType string) bool {
if len(d.Credentials) == 0 {
return false
}
switch connectorType {
case ConnectorTypeRSS:
raw, ok := d.Credentials["auth_headers"]
if !ok {
return false
}
s, ok := raw.(string)
return ok && strings.TrimSpace(s) != ""
default:
return len(d.Credentials) > 0
}
}
// StripNonSecretCredentials removes non-secret values mistakenly stored in the
// credentials map before persistence.
func (d *DataSourceConfig) StripNonSecretCredentials(connectorType string) {
if d == nil || d.Credentials == nil {
return
}
switch connectorType {
case ConnectorTypeRSS:
delete(d.Credentials, "feed_urls")
if len(d.Credentials) == 0 {
d.Credentials = nil
}
}
}
// Resource represents a syncable resource (document, folder, space) from external system
type Resource struct {
// Unique identifier in the external system
+25
View File
@@ -116,3 +116,28 @@ func TestDataSource_ParseConfig_MissingKeyDegradesGracefully(t *testing.T) {
assert.False(t, parsed.HasCredentials() && parsed.Credentials["token"] != "",
"HasCredentials should not report a usable credential when decrypt failed")
}
func TestDataSourceConfig_HasConfiguredCredentials_RSS(t *testing.T) {
feedOnly := DataSourceConfig{
Credentials: map[string]interface{}{
"feed_urls": "https://example.com/feed.xml",
},
}
assert.False(t, feedOnly.HasConfiguredCredentials(ConnectorTypeRSS))
assert.True(t, feedOnly.HasCredentials())
withAuth := DataSourceConfig{
Credentials: map[string]interface{}{
"feed_urls": "https://example.com/feed.xml",
"auth_headers": "Authorization: Bearer x",
},
}
assert.True(t, withAuth.HasConfiguredCredentials(ConnectorTypeRSS))
t.Run("strip feed_urls", func(t *testing.T) {
cfg := feedOnly
cfg.StripNonSecretCredentials(ConnectorTypeRSS)
assert.Nil(t, cfg.Credentials)
assert.False(t, cfg.HasCredentials())
})
}